lambda-serverless/lambda_function.py at master · czam01/lambda-serverless · GitHub

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
importboto3
importjson
importos
importlogging
fromboto3.dynamodb.conditionsimportKey

# CAMBIOOOOOOOOO

logger=logging.getLogger()
logger.setLevel(os.environ.get('LOG_LEVEL', 'INFO'))

DYNAMO_BD=os.environ['DYNAMO_BD']
DYNAMO_KEY=os.environ.get('DYNAMO_KEY', 'cc')

dynamodb=boto3.resource('dynamodb')
table=dynamodb.Table(DYNAMO_BD)


deflambda_handler(event, context):
try:
body=eventifisinstance(event, dict) andDYNAMO_KEYineventelsejson.loads(event.get('body', '{}'))
cc=body.get(DYNAMO_KEY)

ifnotcc:
returnresponse(400, {'error': f'Campo "{DYNAMO_KEY}" es requerido'})

result=table.query(KeyConditionExpression=Key(DYNAMO_KEY).eq(str(cc)))
items=result.get('Items', [])

ifnotitems:
returnresponse(404, {'error': 'Registro no encontrado'})

returnresponse(200, items[0])

exceptjson.JSONDecodeError:
returnresponse(400, {'error': 'JSON inválido en el body'})
exceptExceptionase:
logger.error(f'Error: {e}')
returnresponse(500, {'error': 'Error interno del servidor'})


defresponse(status_code, body):
return {
'statusCode': status_code,
'headers': {'Content-Type': 'application/json'},
'body': json.dumps(body, default=str)
 }