faster_sam package

Subpackages

• faster_sam.cache package
• faster_sam.middlewares package

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,

parameters=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

parameters

Dictionary representing parameters name and default value for CloudFormation deployment.

Type:

Optional[Dict[str, str]]

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_bucket(

app,

)

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

Parameters:

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

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.

configure_schedule(

app,

)

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

Parameters:

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

lambda_mapper(

gateway_id=None,

event_type=EventType.API,

)

Generate a route map extracted from the lambda functions events.

Parameters:

• gateway_id (Optional[str]) – Optional gateway id to filter the routes for a specific API Gateway.

• event_type (EventType) – The type of events to look for.

Returns:

Dictionary containing the routes.

Return type:

Dict[str, Any]

load_environment()

Loads environment variables from CloudFormationTemplate.

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

class faster_sam.cloudformation.Api(

resource_id,

resource,

)

Bases: Resource

property name

property stage_name

class faster_sam.cloudformation.ApiEvent(

resource_id,

resource,

)

Bases: EventSource

property method

property path

property rest_api_id

class faster_sam.cloudformation.Bucket(

resource_id,

resource,

)

Bases: Resource

property name

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,

parameters=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.

• parameters (Optional[Dict[str, str]]) – Dictionary representing parameters name and default value.

template

Dictionary representing the loaded CloudFormation template.

Type:

Dict[str, Any]

property apis

Dict[str, Api]: Dictionary containing API Gateway resources in the CloudFormation template.

property buckets

Dict[str, Bucket]: Dictionary containing buckets resources in the CloudFormation template.

property environment

Dict[str, Any]: Dictionary containing environment variables in the CloudFormation template.

find_environment()

Reads the CloudFormation template to extract environment variables defined at both global and function levels.

Returns:

Dictionary containing environment variables in the CloudFormation template.

Return 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, Function]: Dictionary containing Lambda function resources in the CloudFormation template.

include_files()

Load external files specified in the CloudFormation template like OpenAPI schema.

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, Queue]: Dictionary containing SQS Queue resources in the CloudFormation template.

set_parameters(

parameters,

)

Set the default value of parameters in the CloudFormation template.

Parameters:

parameters (Optional[Dict[str, str]]) – Dictionary representing parameters name and default value.

class faster_sam.cloudformation.EventSource(

resource_id,

resource,

)

Bases: Resource

classmethod from_resource(

resource_id,

resource,

)

property type

class faster_sam.cloudformation.EventType(

value,

names=None,

*values,

module=None,

qualname=None,

type=None,

start=1,

boundary=None,

)

Bases: Enum

Enum representing different types of AWS resource events.

API

Represents the “Api” node type.

Type:

str

SQS

Represents the “SQS” node type.

Type:

str

SCHEDULE

Represents the “Schedule” node type.

Type:

str

S3

Represents the “S3” node type.

Type:

str

API = 'Api'

S3 = 'S3'

SCHEDULE = 'Schedule'

SQS = 'SQS'

class faster_sam.cloudformation.Function(

resource_id,

resource,

)

Bases: Resource

property environment

Returns a dictionary containing the environment variables for the Lambda Function.

Returns:

The environment variables for the Lambda Function.

Return type:

Dict[str, str]

property events

filtered_events(

event_type,

)

property handler

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.

Returns:

The constructed Lambda handler path.

Return type:

str

property name

class faster_sam.cloudformation.IntrinsicFunctions

Bases: object

Resolve intrinsic functions in CloudFormation

static base64(

value,

)

Encode a string to base64.

Parameters:

value (str) – The string to be encoded to base64.

Returns:

The base64-encoded string.

Return type:

str

static eval(

function,

template,

)

Try to resolve an intrinsic function.

Parameters:

• function (Dict[str, Any]) – The intrinsic function and its arguments.

• template (Dict[str, Any]) – A dictionary representing the CloudFormation template.

Returns:

The result of the intrinsic function, or None if it cannot access the value.

Return type:

Any

Raises:

NotImplementedError – If the intrinsic function is not implemented.

static find_in_map(

value,

template,

)

Gets a value from a mapping declared in the CloudFormation template.

Parameters:

• value (List[Any]) – List containing the map name, top-level key, and second-level key.

• template (Dict[str, Any]) – A dictionary representing the CloudFormation template.

Returns:

The value from the map, or None if the map or keys are not found.

Return type:

Any

static get_att(

value,

template,

)

Gets the value of an attribute from a CloudFormation template based on a list of logical name and attribute name.

Parameters:

• value (List[Any]) – List containing the logical name and attribute name

• template (Dict[str, Any]) – A dictionary representing the CloudFormation template.

Returns:

The value of atribute name, or None if the keys are not found.

Return type:

Optional[str]

static join(

value,

template,

)

Joins elements in a list with a specified delimiter.

Parameters:

• value (List[Any]) – A list containing two elements: the delimiter as the first element, and the values to join as the second element.

• template (Dict[str, Any]) – A dictionary representing the CloudFormation template.

Returns:

The joined string if successful; otherwise, None.

Return type:

Optional[str]

static ref(

value,

template,

)

Gets a referenced value from the CloudFormation template.

Parameters:

• value (str) – The name of the referenced value to retrieve.

• template (Dict[str, Any]) – A dictionary representing the CloudFormation template.

Returns:

The referenced value, or None if the reference is not found.

Return type:

Optional[str]

static select(

value,

template,

)

Selects a value from a list based on the given index. If the value at the index is a dictionary, it evaluates it using CloudFormation template data. :param value: A list containing values from which to select. :type value: List[Any] :param template: A dictionary representing the CloudFormation template. :type template: Dict[str, Any]

Returns:

The selected value from the list, or None if any of the evaluated values are None.

Return type:

Optional[str]

static split(

value,

template,

)

Splits a list of values using a specified delimiter.

Parameters:

• value (List[Any]) – A tuple containing the delimiter as its first element, followed by a list of values to split.

• template (Dict[str, Any]) – A dictionary representing the CloudFormation template.

Returns:

A list of strings resulting from splitting using the delimiter. or None if any of the evaluated values are None.

Return type:

Optional[str]

class faster_sam.cloudformation.Queue(

resource_id,

resource,

)

Bases: Resource

property message_retention_period

property name

property redrive_policy

property visibility_timeout

class faster_sam.cloudformation.Resource(

resource_id,

resource,

)

Bases: object

class faster_sam.cloudformation.ResourceType(

value,

names=None,

*values,

module=None,

qualname=None,

type=None,

start=1,

boundary=None,

)

Bases: Enum

Enum representing different types of AWS resources.

API

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

Type:

str

FUNCTION

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

Type:

str

QUEUE

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

Type:

str

BUCKET

Represents the “AWS::S3::Bucket” node type.

Type:

str

API = 'AWS::Serverless::Api'

BUCKET = 'AWS::S3::Bucket'

FUNCTION = 'AWS::Serverless::Function'

QUEUE = 'AWS::SQS::Queue'

class faster_sam.cloudformation.S3Event(

resource_id,

resource,

)

Bases: EventSource

property bucket

class faster_sam.cloudformation.SQSEvent(

resource_id,

resource,

)

Bases: EventSource

property batch_size

property queue

class faster_sam.cloudformation.ScheduleEvent(

resource_id,

resource,

)

Bases: EventSource

property schedule

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.Bucket(

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 bucket

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

Returns:

An bucket 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]

class faster_sam.lambda_event.Schedule(

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 schedule

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

Returns:

An schedule 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.BucketRoute(

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.

class faster_sam.routing.ScheduleRoute(

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