This commit is contained in:
2025-04-12 21:04:02 -03:00
parent 1b2ebcfb99
commit 86bdb41216
19 changed files with 259 additions and 60 deletions

View File

@@ -12,7 +12,6 @@ from uuid import UUID
from aws_lambda_powertools import Logger
from boto3.dynamodb.types import TypeDeserializer, TypeSerializer
from botocore.exceptions import ClientError
from glom import glom
from .dateutils import now, timestamp
from .funcs import omit
@@ -129,12 +128,40 @@ if TYPE_CHECKING:
@dataclass
class SortKey(str):
"""
SortKey encapsulates the sort key value and optionally stores additional attributes
for nested data extraction.
Parameters
----------
sk: str
The sort key value.
table_name: str, optional
Optional name of the table associated with the sort key.
path_spec: str, optional
Optional specification for nested data extraction.
"""
sk: str
table_name: str | None = None
path_spec: str | None = None
else:
class SortKey(str):
"""
SortKey encapsulates the sort key value and optionally stores additional attributes
for nested data extraction.
Parameters
----------
sk: str
The sort key value.
table_name: str, optional
Optional name of the table associated with the sort key.
path_spec: str, optional
Optional specification for nested data extraction.
"""
def __new__(
cls,
sk: str,
@@ -574,6 +601,27 @@ class PaginatedResult(TypedDict):
class DynamoDBCollection:
"""
DynamoDBCollection provides a high-level abstraction for performing common CRUD operations
and queries on a DynamoDB table. It leverages an underlying persistence layer to handle
serialization and deserialization of data, key composition, transaction operations, and TTL management.
This collection class simplifies interaction with DynamoDB items, allowing users to:
- Retrieve a single item or multiple items via transactions.
- Insert (put) items with optional TTL (time-to-live) settings.
- Delete items based on keys and conditions.
- Query items using partition keys or composite key pairs with optional filtering and pagination.
Parameters
----------
persistence_layer: DynamoDBPersistenceLayer
The persistence layer instance responsible for direct DynamoDB operations.
exception_cls: Type[Exception], optional
The exception class to be raised when a requested item is not found.
tz: str, optional
The timezone identifier used for date/time operations.
"""
def __init__(
self,
persistence_layer: DynamoDBPersistenceLayer,
@@ -781,6 +829,8 @@ class DynamoDBCollection:
head, tail = {}, items
def _getin(sk: SortKey, v: dict) -> dict:
from glom import glom
v = omit((PK, SK), v)
return glom(v, sk.path_spec) if sk.path_spec else v

View File

@@ -165,9 +165,11 @@ if TYPE_CHECKING:
CnpjStr = Annotated[str, ...]
else:
class CpfStr(CpfCnpj): ...
class CpfStr(CpfCnpj):
pass
class CnpjStr(CpfCnpj): ...
class CnpjStr(CpfCnpj):
pass
if __name__ == '__main__':

View File

@@ -7,14 +7,56 @@ def pick(
exclude_none: bool = True,
default: Any = None,
) -> dict[str, Any]:
"""Returns a partial copy of an object containing only the keys specified."""
return {
k: dct.get(k, default) for k in keys if k in dct or not exclude_none
}
"""
Return a partial dict with only the specified keys.
Parameters
----------
keys: list[str] or tuple[str, ...]
Keys to select.
dct: dict[str, Any]
Source dict.
exclude_none: bool, optional
If True, omit keys not in dct; if False, include them with `default`.
default: Any, optional
Value for keys not in dct when exclude_none is False.
Returns
-------
dict[str, Any]
A dict with the picked key/value pairs.
Examples
--------
>>> d = {'a': 1, 'b': 2, 'c': 3}
>>> pick(['a', 'c'], d)
{'a': 1, 'c': 3}
>>> pick(['a', 'd'], d, exclude_none=False, default='missing')
{'a': 1, 'd': 'missing'}
"""
return {k: dct.get(k, default) for k in keys if k in dct or not exclude_none}
def omit(
keys: list[str] | tuple[str, ...], dct: dict[str, Any]
) -> dict[str, Any]:
"""Returns a partial copy of an object omitting the keys specified."""
def omit(keys: list[str] | tuple[str, ...], dct: dict[str, Any]) -> dict[str, Any]:
"""
Return a partial dict omitting the specified keys.
Parameters
----------
keys: list[str] or tuple[str, ...]
Keys to omit.
dct: dict[str, Any]
Source dict.
Returns
-------
dict[str, Any]
A dict without the omitted key/value pairs.
Examples
--------
>>> d = {'a': 1, 'b': 2, 'c': 3}
>>> omit(['b'], d)
{'a': 1, 'c': 3}
"""
return {k: dct[k] for k in dct.keys() if k not in keys}