50 lines
1.4 KiB
Python
50 lines
1.4 KiB
Python
from typing import Annotated
|
|
|
|
from aws_lambda_powertools.event_handler.api_gateway import Router
|
|
from aws_lambda_powertools.event_handler.openapi.params import Path
|
|
from layercake.dynamodb import DynamoDBPersistenceLayer, KeyPair, SortKey, TransactKey
|
|
from layercake.extra_types import CnpjStr, CpfStr
|
|
from layercake.funcs import pick
|
|
from pydantic import EmailStr
|
|
|
|
from boto3clients import dynamodb_client
|
|
from config import OAUTH2_TABLE
|
|
|
|
router = Router()
|
|
dyn = DynamoDBPersistenceLayer(OAUTH2_TABLE, dynamodb_client)
|
|
|
|
|
|
@router.get('/lookup')
|
|
def lookup(
|
|
email: Annotated[EmailStr, Path] = 'unknown',
|
|
cpf: Annotated[CpfStr, Path] = 'unknown',
|
|
cnpj: Annotated[CnpjStr, Path] = 'unknown',
|
|
):
|
|
r = dyn.collection.get_items(
|
|
KeyPair(
|
|
pk='email',
|
|
sk=SortKey(email, path_spec='user_id'),
|
|
rename_key='id',
|
|
)
|
|
+ KeyPair(
|
|
pk='cpf',
|
|
sk=SortKey(cpf, path_spec='user_id'),
|
|
rename_key='id',
|
|
)
|
|
+ KeyPair(
|
|
pk='cnpj',
|
|
sk=SortKey(cnpj, path_spec='user_id'),
|
|
rename_key='org_id',
|
|
),
|
|
flatten_top=False,
|
|
)
|
|
|
|
if 'id' in r:
|
|
user = dyn.collection.get_items(
|
|
TransactKey(r['id']) + SortKey('0') + SortKey('FRESH_USER')
|
|
)
|
|
|
|
return r | pick(('name', 'email'), user) if 'FRESH_USER' in user else {}
|
|
|
|
return r
|