faster_sam package¶

Subpackages¶

Submodules¶

faster_sam.adapter module¶

exception faster_sam.adapter.GatewayLookupError¶

Bases: LookupError

Raised when performing API Gateway lookup, usually when multiple gateways are available requiring an explicit target.

class faster_sam.adapter.SAM( template_path=None, )¶

Bases: object

Adapter class for FastAPI applications allowing map API routes defined on a CloudFormation template or OpenAPI file.

Parameters:: template_path (Optional[str]) – Path to the CloudFormation template file.

template¶

Instance of CloudformationTemplate based on the provided template_path.

Type:: CloudformationTemplate

configure_api( app, gateway_id=None, )¶

Configures the FastAPI application with routes based on one of the following template files: CloudFormation, OpenAPI.

Parameters:

app (FastAPI) – The FastAPI application instance to be configured.
gateway_id (Optional[str], optional) – Optional API Gateway resource ID defined on the CloudFormation template. When multiple API Gateways are defined on the template this argument is required to identify which gateway is being mapped.

Raises:

GatewayLookupError – Raised if multiple API Gateways are defined on the CloudFormation template and the gateway_id are not set.

configure_queues( app, )¶

Configures the FastAPI application with routes based on queues defined in cloudformation template.

Parameters:: app (FastAPI) – The FastAPI application instance to be configured.

lambda_mapper( gateway_id, )¶

Generate a route map extracted from the lambda functions schema corresponding to the given gateway id.

Parameters:: gateway_id (Optional[str]) – Optional gateway id to filter the routes for a specific API Gateway.
Returns:: Dictionary containing the routes.
Return type:: Dict[str, Any]

lambda_queue_mapper()¶

Generate a route map extracted from the lambda functions that is a queue consumer using the name of the queue as path.

Returns:: Dictionary containing the routes.
Return type:: Dict[str, Any]

openapi_mapper( openapi_schema, )¶

Create a route map extracted from the given OpenAPI schema.

Notice that this is an OpenAPI schema used by AWS SAM to configure an API Gateway, therefore it requires AWS extensions pointing to event consumers, most of the time Lambda Functions.

e.g.

>>> # The OpenAPI input schema is omitting information for brevity
>>> schema = '''
... paths:
...   /health:
...     get:
...       x-amazon-apigateway-integration:
...          passthroughBehavior: when_no_match
...          httpMethod: POST
...          type: aws_proxy
...          uri:
...            Fn::Sub: "arn:aws:apigateway::lambda:path/2015-03-31/
...            functions/${HealthFunction.Arn}/invocations"
...          credentials: "arn:aws:iam:::role/apigateway-invoke-lambda-role"
... '''
>>> SAM().openapi_mapper(schema)
{"/health": {"get": {"handler": "src.handlers.health"}}}

Parameters:: openapi_schema (Dict[str, Any]) – OpenAPI schema dictionary.
Returns:: Dictionary containing the routes.
Return type:: Dict[str, Any]

register_routes( app, routes, class_override=None, )¶

Registers FastAPI routes based on the provided route map into the given FastAPI application.

Parameters:

app (FastAPI) – The FastAPI application instance.
routes (Dict[str, Any]) – Dictionary containing FastAPI routes.

faster_sam.cloudformation module¶

exception faster_sam.cloudformation.CFBadNode¶

Bases: ValueError

Raised when an invalid CloudFormation node is encountered.

exception faster_sam.cloudformation.CFBadTag¶

Bases: TypeError

Raised when an invalid CloudFormation tag is encountered.

class faster_sam.cloudformation.CFLoader( stream, )¶

Bases: SafeLoader

Custom YAML loader for CloudFormation templates.

yaml_multi_constructors = {'!': <function multi_constructor>}¶

exception faster_sam.cloudformation.CFTemplateNotFound¶

Bases: FileNotFoundError

Raised when the CloudFormation template file cannot be found.

class faster_sam.cloudformation.CloudformationTemplate( template_path=None, )¶

Bases: object

Represents an AWS CloudFormation template and provides methods for extracting information from the template.

Parameters:: template_path (Optional[str]) – Path to the CloudFormation template file.

template¶

Dictionary representing the loaded CloudFormation template.

Type:: Dict[str, Any]

find_nodes( tree, node_type, )¶

Finds nodes of a specific type in the CloudFormation template.

Parameters:

tree (Dict[str, Any]) – Dictionary representing a subset of the CloudFormation template.
node_type (NodeType) – The type of node to search for.

Returns:

Dictionary containing nodes of the specified type.

Return type:

Dict[str, Any]

property functions¶: Dict[str, Any]: Dictionary containing Lambda function resources in the CloudFormation template.

property gateways¶: Dict[str, Any]: Dictionary containing API Gateway resources in the CloudFormation template.

include_files()¶: Load external files specified in the CloudFormation template like OpenAPI schema.

lambda_handler( resource_id, )¶

Returns a string representing the full module path for a Lambda Function handler. The path is built by joining the code URI and the handler attributes on the CloudFormation for the given Lambda Function identified by resource_id. :param resource_id: The id of the Lambda function resource. :type resource_id: str

Returns:: The constructed Lambda handler path.
Return type:: str

load( template=None, )¶

Reads CloudFormation template file from the disk and convert it to a dictionary.

If the template argument is not set it is assumed an YAML file named template exists in the current directory.

Parameters:: template (Optional[str]) – Path to the CloudFormation template file.
Returns:: Dictionary representing the loaded CloudFormation template.
Return type:: Dict[str, Any]
Raises:: CFTemplateNotFound – Exception raised when CloudFormation template file is not found.

property queues¶: Dict[str, Any]: Dictionary containing SQS Queue resources in the CloudFormation template.

class faster_sam.cloudformation.NodeType( value, names=None, *values, module=None, qualname=None, type=None, start=1, boundary=None, )¶

Bases: Enum

Enum representing different types of CloudFormation nodes.

API_GATEWAY¶

Represents the “AWS::Serverless::Api” node type.

Type:: str

LAMBDA¶

Represents the “AWS::Serverless::Function” node type.

Type:: str

QUEUE¶

Represents the “AWS::SQS::Queue” node type.

Type:: str

API_EVENT¶

Represents the “Api” node type.

Type:: str

API_EVENT = 'Api'¶

API_GATEWAY = 'AWS::Serverless::Api'¶

LAMBDA = 'AWS::Serverless::Function'¶

QUEUE = 'AWS::SQS::Queue'¶

SQS_EVENT = 'SQS'¶

faster_sam.cloudformation.construct_getatt( node, )¶

Custom YAML node constructor for AWS CloudFormation GetAtt intrinsic function.

Parameters:: node (yaml.nodes.Node) – The node representing a CloudFormation GetAtt element.
Returns:: List representing the constructed CloudFormation GetAtt element.
Return type:: List[Any]
Raises:: CFBadNode – If an invalid CloudFormation node is encountered.

faster_sam.cloudformation.multi_constructor( loader, tag_suffix, node, )¶

Custom YAML node constructor.

Handles AWS CloudFormation extensions for its short version of intrinsic functions.

Parameters:

loader (CFLoader) – The YAML loader instance.
tag_suffix (str) – The YAML tag suffix.
node (yaml.nodes.Node) – The YAML node.

Returns:

A dictionary representation of the given YAML node.

Return type:

Dict[str, Any]

Raises:

CFBadTag – If an invalid CloudFormation tag is encountered.

faster_sam.lambda_event module¶

class faster_sam.lambda_event.ApiGateway( request, endpoint, )¶

Bases: ResourceInterface

async call_endpoint()¶

Call event buider and retuns a custom response based on the mapped event.

Returns:: A response object.
Return type:: Response

async event_builder()¶

Builds an event of type aws_proxy from API Gateway.

It uses the given request object to fill the event details.

Returns:: An aws_proxy event.
Return type:: Dict[str, Any]

class faster_sam.lambda_event.CustomResponse( data, )¶

Bases: Response

Represents an HTTP response.

class faster_sam.lambda_event.ResourceInterface( request, endpoint, )¶

Bases: ABC

Interface for aws resources.

This abstract base class defines an interface for aws resources to implement. Subclasses must implement the call_endpoint and event_builder methods to map events.

abstract async call_endpoint()¶

Call event buider and retuns a custom response based on the mapped event.

Returns:: A response object.
Return type:: Response

abstract async event_builder()¶

Builds an event based on the current request.

Returns:: The mapped event.
Return type:: Dict[str, Any]

class faster_sam.lambda_event.SQS( request, endpoint, )¶

Bases: ResourceInterface

async call_endpoint()¶

Call event buider and retuns a custom response based on the mapped event.

Returns:: A response object.
Return type:: Response

async event_builder()¶

Builds an event of type sqs

It uses the given request object to fill the event details.

Returns:: An sqs event.
Return type:: Dict[str, Any]

faster_sam.openapi module¶

faster_sam.openapi.custom_openapi( app, openapi_schema, )¶

Returns an OpenAPI schema generator function for FastAPI.

The generator function uses the given input schema as the base to generate a new schema combining it with the auto-generated by FastAPI.

e.g

>>> with open("swagger.json") as fp:
...     schema = json.load(fp)
...
>>> app = FastAPI()
>>> app.openapi = custom_openapi(app, schema)

Parameters:

app (FastAPI) – A FastAPI application instance.
openapi_schema (Dict[str, Any]) – OpenAPI schema to be merged with the new one.

Returns:

A callable function that generates the OpenAPI schema.

Return type:

Callable[[], Dict[str, Any]]

faster_sam.routing module¶

class faster_sam.routing.APIRoute(

path,

endpoint,

*args,

**kwargs,

)¶

Bases: APIRoute

Extends FastAPI Router class used to describe path operations. This custom router class receives the endpoint parameter as a string with the full module path instead of the actual callable.

class faster_sam.routing.QueueRoute(

path,

endpoint,

*args,

**kwargs,

)¶

Bases: APIRoute

Extends FastAPI Router class used to describe path operations. This custom router class receives the endpoint parameter as a string with the full module path instead of the actual callable.

faster_sam.routing.handler( func, Resource, )¶

Returns a wrapper function.

The returning function converts a request object into a event, then the event is passed to the handler function, finally the function result is converted to a response object.

Parameters:: func (Handler) – A callable object.
Returns:: An async function, which accepts a single request argument and return a response.
Return type:: Endpoint

faster_sam.routing.import_handler( path, )¶

Returns a callable object from the given module path.

Parameters:: path (str) – Full module path.
Returns:: A callable object.
Return type:: Handler

faster_sam.web_identity_providers module¶

class faster_sam.web_identity_providers.GCPProvider¶

Bases: ProviderInterface

Requests identity information from the metadata server.

e.g

This example retrieves an identity token in the JWT format.

>>> provider = GCPProvider()
>>> provider.get_token()

Environment Variables¶

WEB_IDENTITY_AUDIENCEstr: The unique URI agreed upon by both the instance and the system verifying the instance’s identity. For example, the audience could be a URL for the connection between the two systems.
WEB_IDENTITY_FORMATstr: The optional parameter that specifies whether the project and instance details are included in the payload. Specify full to include this information in the payload or standard to omit the information from the payload. The default value is standard.
WEB_IDENTITY_LICENSESstr: An optional parameter that specifies whether license codes for images associated with this instance are included in the payload. Specify TRUE to include this information or FALSE to omit this information from the payload. The default value is FALSE. Has no effect unless format is full.

get_token()¶

Get a JSON Web Token (JWT) signed by Google using the RS256 algorithm containing the identity information of the requester.

Returns:: token – The identity token if available, else None.
Return type:: str or None

class faster_sam.web_identity_providers.Provider( value, names=None, *values, module=None, qualname=None, type=None, start=1, boundary=None, )¶

Bases: Enum

Enumeration of identity providers.

Providers:: GCP: Google Cloud Platform identity provider.

GCP = 'gcp'¶

class faster_sam.web_identity_providers.ProviderInterface¶

Bases: ABC

Interface for identity providers.

This abstract base class defines an interface for identity providers to implement. Subclasses must implement the get_token method to provide functionality for retrieving identity tokens.

abstract get_token()¶

Get the identity token.

This method should be implemented by subclasses to retrieve the identity token from the respective identity provider.

Returns:: token – The identity token if available, else None.
Return type:: str or None

faster_sam.web_identity_providers.factory( provider, )¶

Factory function for creating identity provider instances.

This function takes a provider name as input and returns an instance of the corresponding identity provider class.

e.g

This example get an instance of provider.

>>> provider = factory(provider="gcp")
>>> token = provider.get_token()

Parameters:: provider (str) – The name of the identity provider.
Returns:: provider – An instance of the identity provider class corresponding to the specified provider name.
Return type:: ProviderInterface

faster_sam package¶

Subpackages¶

Submodules¶

faster_sam.adapter module¶

faster_sam.cloudformation module¶

faster_sam.lambda_event module¶

faster_sam.openapi module¶

faster_sam.routing module¶

faster_sam.web_identity_providers module¶

Environment Variables¶

Faster SAM

Navigation

Related Topics