saladeaula.digital/http-api/app/rules/user.py

from types import SimpleNamespace
from typing import TypedDict

from aws_lambda_powertools.event_handler.exceptions import (
    BadRequestError,
)
from layercake.dateutils import now, ttl
from layercake.dynamodb import (
    ComposeKey,
    DynamoDBPersistenceLayer,
    KeyPair,
)

User = TypedDict('User', {'id': str, 'name': str, 'cpf': str})


def update_user(
    data: User,
    /,
    *,
    persistence_layer: DynamoDBPersistenceLayer,
) -> bool:
    now_ = now()
    user = SimpleNamespace(**data)
    # Get the user's CPF, if it exists.
    old_cpf = persistence_layer.get_item(KeyPair(user.id, '0')).get('cpf', None)

    with persistence_layer.transact_writer() as transact:
        transact.update(
            key=KeyPair(user.id, '0'),
            update_expr='SET #name = :name, cpf = :cpf, update_date = :update_date',
            expr_attr_names={
                '#name': 'name',
            },
            expr_attr_values={
                ':name': user.name,
                ':cpf': user.cpf,
                ':update_date': now_,
            },
            cond_expr='attribute_exists(sk)',
        )
        # Prevent the user from updating more than once every 24 hours
        transact.put(
            item={
                'id': user.id,
                'sk': 'last_profile_edit',
                'create_date': now_,
                'ttl': ttl(start_dt=now_ + timedelta(hours=24)),
            },
            cond_expr='attribute_not_exists(sk)',
        )

        class CPFConflictError(BadRequestError):
            def __init__(self, msg: str):
                super().__init__('Cpf already exists')

        if user.cpf != old_cpf:
            transact.put(
                item={
                    'id': 'cpf',
                    'sk': user.cpf,
                    'user_id': user.id,
                    'create_date': now_,
                },
                cond_expr='attribute_not_exists(sk)',
                exc_cls=CPFConflictError,
            )

        # Ensures that the old CPF is discarded
        if old_cpf:
            transact.delete(key=KeyPair('cpf', old_cpf))

    return True


def add_email(
    id: str,
    email: str,
    /,
    *,
    persistence_layer: DynamoDBPersistenceLayer,
):
    now_ = now()

    with persistence_layer.transact_writer() as transact:
        transact.update(
            key=KeyPair(id, '0'),
            update_expr='ADD emails :email',
            expr_attr_values={
                ':email': {email},
            },
        )
        transact.put(
            item={
                'id': id,
                'sk': f'emails#{email}',
                'email_primary': False,
                'email_verified': False,
                'create_date': now_,
            },
            cond_expr='attribute_not_exists(sk)',
        )

        class EmailConflictError(BadRequestError):
            def __init__(self, msg: str):
                super().__init__('Email already exists')

        transact.put(
            item={
                'id': 'email',
                'sk': email,
                'user_id': id,
                'create_date': now_,
            },
            cond_expr='attribute_not_exists(sk)',
            exc_cls=EmailConflictError,
        )

    return True


def del_email(
    id: str,
    email: str,
    /,
    *,
    persistence_layer: DynamoDBPersistenceLayer,
) -> bool:
    """Delete any email except the primary email."""
    with persistence_layer.transact_writer() as transact:
        transact.delete(
            key=KeyPair('email', email),
        )
        transact.delete(
            key=KeyPair(id, ComposeKey(email, prefix='emails')),
            cond_expr='email_primary <> :primary',
            expr_attr_values={':primary': True},
            exc_cls=BadRequestError,
        )
        transact.update(
            key=KeyPair(id, '0'),
            update_expr='DELETE emails :email',
            expr_attr_values={
                ':email': {email},
            },
        )
    return True


def set_email_as_primary(
    id: str,
    new_email: str,
    old_email: str,
    /,
    *,
    email_verified: bool = False,
    persistence_layer: DynamoDBPersistenceLayer,
):
    now_ = now()
    expr = 'SET email_primary = :email_primary, update_date = :update_date'

    with persistence_layer.transact_writer() as transact:
        # Set the old email as non-primary
        transact.update(
            key=KeyPair(id, ComposeKey(old_email, 'emails')),
            update_expr=expr,
            expr_attr_values={
                ':email_primary': False,
                ':update_date': now_,
            },
        )
        # Set the new email as primary
        transact.update(
            key=KeyPair(id, ComposeKey(new_email, 'emails')),
            update_expr=expr,
            expr_attr_values={
                ':email_primary': True,
                ':update_date': now_,
            },
        )
        transact.update(
            key=KeyPair(id, '0'),
            update_expr='SET email = :email, email_verified = :email_verified, \
                             update_date = :update_date',
            expr_attr_values={
                ':email': new_email,
                ':email_verified': email_verified,
                ':update_date': now_,
            },
        )

    return True


def del_org_member(
    id: str,
    *,
    org_id: str,
    persistence_layer: DynamoDBPersistenceLayer,
) -> bool:
    with persistence_layer.transact_writer() as transact:
        # Remove the user's relationship with the organization and their privileges
        transact.delete(key=KeyPair(id, f'acls#{org_id}'))
        transact.delete(key=KeyPair(id, f'orgs#{org_id}'))
        transact.update(
            key=KeyPair(id, '0'),
            update_expr='DELETE #tenant :org_id',
            expr_attr_names={'#tenant': 'tenant__org_id'},
            expr_attr_values={':org_id': {org_id}},
        )

        # Remove the user from the organization's admins and members list
        transact.delete(key=KeyPair(org_id, f'admins#{id}'))
        transact.delete(key=KeyPair(f'orgmembers#{org_id}', id))

    return True