GitLocker: The Coding Marketplace

Description:

pythonserverlesscrud 1.4.1

python-serverless-crud
The idea
Simple and powerful tool for quick serverless data management via API.
Key concepts

Don't Repeat Yourself - easy model definition with schema and cloud formation generation support
Best practices applied by default (created with AWS LambdaPower Tools)
Flexibility - enable, extend and modify what is needed
One ring to rule them all - support for REST API, GraphQL (via API Gateway), AppSync GraphQL (direct resolvers)

Features

Full CRUD support with validation
Native support for DynamoDB (including CloudFormation creation via troposphere)

GlobalSecondaryIndex support
LocalSecondaryIndex support
Primary Key with and without sort keys

Support for Scan, Query operations on the tables and indexes
Virtual List method on the table or index
Integrated record owner feature with KeyCondition and FilterCondition support (auto-detect)

Documentation
Sample service
from aws_lambda_powertools import Tracer
from aws_lambda_powertools.logging import correlation_paths
from serverless_crud import api
from serverless_crud.dynamodb import annotation as db
from serverless_crud.model import BaseModel
from serverless_crud.logger import logger

tracer = Tracer()

@db.Model(
key=db.PrimaryKey(id=db.KeyFieldTypes.HASH),
indexes=(
db.GlobalSecondaryIndex("by_user", user=db.KeyFieldTypes.HASH, created=db.KeyFieldTypes.RANGE),
),
owner_field="user"
)
class Device(BaseModel):
id: str
created: int
user: str = None

api.rest.registry(Device, alias="device")

@logger.inject_lambda_context(correlation_id_path=correlation_paths.API_GATEWAY_REST)
@tracer.capture_lambda_handler
def rest_handler(event, context):
return api.rest.handle(event, context)

With just a few lines of the code we are able to create Device model service which then can be extended.
In this example we:

Defined our Device model with some extra metadata, used by our generators. That includes:

Table key definition
GlobalSecondaryIndex
Definition of the field which will hold owner of the record (identity provided by cognito)

Registered our Device model into rest API under device alias
Created rest handler which then can be referred in our serverless.yml file

A few notes here:

we need to define rest_handler function if we would like to use it as a target for local execution with serverless freamework
Lambda Power Tools are build around functions and they don't work properly with object methods
We use one function per API type, and we relay on internal router provided by each API implementation

Serverless integration
If you use (serverless-builder)[https://github.com/epsyhealth/serverless-builder] you can create your serverless.yml with just a few lines of code (including DynamodbTables)
from serverless import Service, Configuration
from serverless.aws.features import XRay
from serverless.aws.functions.http import HTTPFunction
from serverless.plugins import PythonRequirements, Prune
from serverless.provider import AWSProvider
from troposphere import dynamodb

from timemachine_api.handlers import api

service = Service(
"timemachine-api",
"Collect events in chronological order",
AWSProvider(),
config=Configuration(
domain="epsy.app"
)
)
service.provider.timeout = 5

service.plugins.add(Prune())
service.plugins.add(PythonRequirements(layer=False, useStaticCache=False, dockerSsh=True))

service.enable(XRay())

for name, table_specification in api.dynamodb_table_specifications().items():
service.resources.add(dynamodb.Table(name, **table_specification))

authorizer = dict(name="auth",
arn="arn:aws:cognito-idp:us-east-1:772962929486:userpool/us-east-1_FCl7gKtHC")

service.builder.function.http("rest", "Time machine REST API", "/rest/{proxy+}", HTTPFunction.ANY,
handler="timemachine_api.handlers.rest_handler", authorizer=authorizer)

service.render()

Internals
Annotations
serverless-crud project provides one annotation which must be used for all managed models.
from serverless_crud.dynamodb import annotation as db
@db.Model(
key=db.PrimaryKey(name=db.KeyFieldTypes.HASH),
indexes=(
db.GlobalSecondaryIndex(...),
db.LocalSecondaryIndex(...)
),
owner_field="field"
)

Model annotation accepts:

key - primary key definition, in form of kwargs where name of parameter would be a field name which should
be used in key, and value should be a value of KeyFieldTypes enum
indexes - list of indexes GlobalSecondaryIndex|LocalSecondaryIndex. Indexes are defined in same way as primary key
owner_field - name of the field which should be used for data filtering (based on the cognito identity)

Data owner
serverless-crud can enforce some base data filtering on all kind of operations using Dynamodb conditional operations.
If you would like to use this feature you must set owner_field on each model you would like to use this feature.
Library will use this field for:

setting its value on model creation / update (it will overwrite any value provided by user)
as an extra ConditionExpression during GET and DELETE operations
as a part of either FilterExpression or KeyExpression for Scan, Query and List operations

Model registration
To be able to manage given model, you must first register it with specific API.
This can be done with a single line of code:
api.rest.registry(Device, alias="device")

You need to provide only a model type to registry method, all other parameters are optional.
If you like, you can omit alias parameter, in that case framework will use model class name.
Customizing endpoint behaviour
Framework defines a set of classes located in serverless_crud.actions:

CreateAction
DeleteAction
GetAction
ScanAction, ListAction, QueryAction
UpdateAction

all those classes are subclasses of serverless_crud.actions.base.Action class and can be extended if needed.
You may need to execute custom logic after object creation, that can be done with custom CreateAction subclass
from serverless_crud.actions import CreateAction

class CreateDeviceAction(CreateAction):
def handle(self, event: APIGatewayProxyEvent, context):
super().handle(event, context)

# custom logic

api.rest.registry(Device, create=CreateDeviceAction)

You can set custom handlers for each supported operation:
def registry(self, model, alias=None, get=GetAction, create=CreateAction, update=UpdateAction, delete=DeleteAction,
lookup_list=ListAction, lookup_scan=ScanAction, lookup_query=QueryAction):

As you can see, all actions are defined by default. That also means that all actions are enabled by default, but
each action can be disabled.
If you need to disable action you need to set action handler to None, that will prevent framework from creating
route for given action, and it will disable whole logic behind it.
Routes
REST API specific feature.
Framework will create multiple routes for each register model, using alias as a URL namespace.
Generated routes:

GET /rest/{alias}/{pk} - fetch object by PK (see notes about PK below)
POST /rest/{alias} - create new record
PUT /rest/{alias}/{pk} - update record with given PK
DELETE /rest/{alias}/{pk} - delete record with given PK
GET /rest/lookup/{alias}/list - list all the records of given type using Query on the table
GET /rest/lookup/{alias}/list/{index_name} - list all the records of the given type using Query on specific index
POST /rest/lookup/{alias}/query - perform a query on given table
POST /rest/lookup/{alias}/query/{index_name} - perform a query on given index
POST /rest/lookup/{alias}/scan - perform a scan on given table
POST /rest/lookup/{alias}/scan/{index_name} - perform a scan on given index

Primary Keys

Please remember that with DynamoDB key is a Partition Key with optional Sort Key.
In case you define Sort Key DynamoDB will require a value for it while getting / deleting key.
In that case framework will modify routes to include sort key as an extra path parameter

Endpoints