aws

flow

cognito

class saigon.aws.cognito.CognitoClient(config)

Bases: CognitoIdp

Client for comprehensive interaction with AWS Cognito (User Pools and Identity Pools).

This class extends CognitoIdp to add functionalities for user authentication (login) and obtaining temporary AWS IAM credentials from Identity Pools using the authenticated user’s ID token.

property get_cognito_url

Returns the Cognito URL string used for authentication in Identity Pool logins.

This URL combines the region and user pool ID, forming the key required by Cognito Identity Pool when associating an ID token with an identity.

Returns:: The formatted Cognito URL.
Return type:: str

get_iam_credentials(id_token)

Obtains temporary AWS IAM credentials for an authenticated user.

This method first gets an Identity ID from the Cognito Identity Pool using the provided ID token, and then exchanges that Identity ID and ID token for temporary AWS IAM credentials.

Parameters:

id_token (str) – The ID token obtained from a successful Cognito User Pool login.

Returns:

A dictionary containing temporary AWS IAM credentials: (AccessKeyId, SecretKey, SessionToken, Expiration).

Return type:

CredentialsTypeDef

login_user(username, password, new_password=None)

Logs in a user to the Cognito User Pool.

This method handles the user password authentication flow. If a NEW_PASSWORD_REQUIRED challenge is returned (e.g., for first-time login with a temporary password), it attempts to respond to that challenge.

Parameters:

username (str) – The username of the user.
password (str) – The user’s password.
new_password (str | None) – A new password required if the user’s password needs to be changed (e.g., for a temporary password). If None and a new password is required, the original password is used again for the new password. Defaults to None.

Returns:

A dictionary containing the authentication result,: including ID, Access, and Refresh tokens if successful.

Return type:

AuthenticationResultTypeTypeDef

class saigon.aws.cognito.CognitoClientConfig(**data)

Bases: CognitoIdpConfig

Configuration for a full Cognito client, extending IdP configuration.

identity_pool_id

The ID of the Cognito Identity Pool.

Type:: str

client_id

The Client ID associated with the User Pool app client.

Type:: str

class saigon.aws.cognito.CognitoIdp(config)

Bases: object

Client for interacting with AWS Cognito Identity Provider (User Pools).

This class provides methods for managing users within a Cognito User Pool, such as creating, deleting, and confirming users.

property aws_region: str

Returns the AWS region associated with this Cognito IdP client.

Returns:: The AWS region name.
Return type:: str

confirm_user(email)

Confirms a user’s sign-up in the Cognito User Pool.

This action is typically used by administrators to bypass the email verification step for a user.

Parameters:: email (str) – The email (which serves as username) of the user to confirm.

create_user(username_or_alias, notify_user=True, self_verify=False, group_name=None, temporary_password=None, extra_user_attrs=mappingproxy({}))

Creates a new user in the Cognito User Pool or retrieves an existing one.

If the user already exists, their UUID is returned along with True for already_exists. Otherwise, a new user is created with the specified attributes. The user can optionally be added to a group and have a temporary password set.

Parameters:

username_or_alias (str) – The username or email alias for the user.
notify_user (bool) – If True, Cognito sends a welcome message to the user. If False, message sending is suppressed. Defaults to True.
self_verify (bool) – If True, the user’s email is marked as verified. Defaults to False.
group_name (str) – The name of the Cognito group to which the user should be added. Defaults to None.
temporary_password (str) – A temporary password for the new user. If None, Cognito generates one (and sends it if notify_user is True). Defaults to None.
extra_user_attrs (Dict) – A dictionary of additional user attributes (e.g., ‘given_name’, ‘family_name’).Defaults to an empty immutable mapping.

Returns:

A tuple where the first element is the UUID of the user,: and the second element is a boolean indicating whether the user already existed (True) or was newly created (False).

Return type:

Tuple[uuid.UUID, bool]

delete_user(username)

Deletes a user from the Cognito User Pool.

Parameters:: username (str) – The username of the user to delete.
Returns:: True if the user was successfully deleted, False if the user was not found.
Return type:: bool

class saigon.aws.cognito.CognitoIdpConfig(**data)

Bases: BaseModel

Configuration for a Cognito Identity Provider (IdP) client.

user_pool_id

The ID of the Cognito User Pool.

Type:: str

region

The AWS region where the User Pool is located. If None, the default boto3 region will be used.

Type:: Optional[str]

class saigon.aws.cognito.CognitoRequestContext(**data): Bases: HeaderContext[TypeVar]

saigon.aws.cognito.get_user_pool_identity_from_iam_auth_provider(iam_auth_provider)

Extracts the user pool identity (UUID) from the ‘X-Cognito-AuthProvider’ header.

This function parses the AWS Cognito IAM Auth Provider header string to isolate and return the UUID representing the user’s identity within the Cognito User Pool.

Parameters:: iam_auth_provider (str) – The value of the ‘X-Cognito-AuthProvider’ header, injected by FastAPI. Expected format is ‘cognito-idp.${REGION}.amazonaws.com/${USER_POOL_ID},cognito-idp.${REGION}.amazonaws.com/${USER_POOL_ID}:CognitoSignIn:${USER_POOL_IDENTITY}’.
Returns:: The UUID of the user pool identity.
Return type:: uuid.UUID

s3

class saigon.aws.s3.S3ObjectDescriptor

Bases: TypedDict

A TypedDict representing the essential identifiers for an S3 object.

Bucket

The name of the S3 bucket.

Type:: str

Key

The key (path) of the object within the bucket.

Type:: str

saigon.aws.s3.s3_get_client()

Returns a Boto3 S3 client instance.

Returns:: A Boto3 S3 client.
Return type:: S3Client

saigon.aws.s3.s3_object_descriptor_from_event(event)

Extracts an S3 object descriptor from an AWS S3 event.

This function parses a standard AWS S3 event payload (e.g., from Lambda) to retrieve the bucket name and object key.

Parameters:

event (S3EventData) – The S3 event data dictionary, typically found within a Lambda event record.

Returns:

A dictionary containing the ‘Bucket’ name and ‘Key’ of the S3 object: involved in the event.

Return type:

S3ObjectDescriptor

saigon.aws.s3.s3_object_to_file(s3_client, **kwargs)

Downloads an S3 object to a temporary local file.

The file is saved within a temporary directory created by this function. The path of the downloaded file is constructed from the temporary directory and the S3 object’s key.

Parameters:

s3_client (S3Client) – The Boto3 S3 client instance.
**kwargs (Unpack[S3ObjectDescriptor]) – Keyword arguments representing an S3ObjectDescriptor, typically containing ‘Bucket’ and ‘Key’.

Returns:

The pathlib.Path object pointing to the downloaded temporary file.

Return type:

Path

saigon.aws.s3.s3_object_unzip(s3_client, **kwargs)

Downloads and unzips an S3 object (assumed to be a ZIP file) to a temporary directory.

The ZIP file is first downloaded using s3_object_to_file, and then its contents are extracted into the same temporary directory as the ZIP file. The original ZIP file itself is excluded from the returned list of paths.

Parameters:

s3_client (S3Client) – The Boto3 S3 client instance.
**kwargs (Unpack[S3ObjectDescriptor]) – Keyword arguments representing an S3ObjectDescriptor, typically containing ‘Bucket’ and ‘Key’ of the ZIP file.

Returns:

A list of pathlib.Path objects for all files extracted from the ZIP archive.

Return type:

List[Path]

saigon.aws.s3.s3_virtual_host_object_url(region, bucket, object_key=None)

Generates a virtual-hosted style URL for an S3 bucket or object.

This format is https://<bucket-name>.s3.<region>.amazonaws.com/<object-key>. If object_key is not provided, it generates a URL for the bucket itself.

Parameters:

region (str) – The AWS region of the S3 bucket.
bucket (str) – The name of the S3 bucket.
object_key (Optional[str]) – The key of the object within the bucket. Defaults to None.

Returns:

The generated virtual-hosted style URL.

Return type:

str

secrets

class saigon.aws.secrets.AwsSecretBaseDbEnv(var_prefix, credentials_type=<class 'saigon.orm.config.PostgreSQLCredentials'>, **kwargs)

Bases: BaseDbEnv

Subclass of BaseDbEnv that provides AwsSecretBaseDbEnv as provider to access the DB secret.

class saigon.aws.secrets.AwsSecretVault(secrets_client=None)

Bases: SecretVault

Concrete implementation of SecretVault that uses boto3 SecretsManager

get_secret(secret_model, secret_key)

Returns the secret specified by its secret_key represented as a SecretModel object.

Parameters:

secret_model (Type[TypeVar(SecretModel, bound= BaseModel)]) – Type representation of the secret content
secret_key (str) – Secret unique identifier

Return type:

TypeVar(SecretModel, bound= BaseModel)

Returns:

Secret data represented as the specified SecretModel

saigon.aws.secrets.get_secret_as_model(model_type, secret_name, secrets_client=None)

Retrieves a secret from AWS Secrets Manager and deserializes it into a Pydantic model.

This function fetches the secret string from Secrets Manager and then uses the provided Pydantic model_type to parse the JSON string into a model instance.

Parameters:

model_type (Type[ModelTypeDef]) – The Pydantic model class to deserialize the secret into.
secret_name (str) – The name or ARN of the secret to retrieve.
secrets_client (Optional[SecretsManagerClient]) – An optional Boto3 Secrets Manager client. If None, a new client will be created. Defaults to None.

Returns:

An instance of the specified Pydantic model populated with the secret’s data.

Return type:

ModelTypeDef

ssm

class saigon.aws.ssm.AwsSsmBaseDbEnv(var_prefix, credentials_type=<class 'saigon.orm.config.PostgreSQLCredentials'>, ssm_client=None, **kwargs)

Bases: BaseDbEnv

Subclass of BaseDbEnv that provides AwsSsmBaseDbEnv as provider to access the DB secret.

class saigon.aws.ssm.AwsSsmVault(ssm_client=None)

Bases: SecretVault

Concrete implementation of SecretVault that uses boto3 SsmClient

get_secret(secret_model, secret_key)

Returns the secret specified by its secret_key represented as a SecretModel object.

Parameters:

secret_model (Type[TypeVar(SecretModel, bound= BaseModel)]) – Type representation of the secret content
secret_key (str) – Secret unique identifier

Return type:

TypeVar(SecretModel, bound= BaseModel)

Returns:

Secret data represented as the specified SecretModel

saigon.aws.ssm.get_parameter_as_model(model_type, parameter, with_decryption=False, ssm_client=None)

Retrieves a parameter from AWS SSM and deserializes it as JSON into a Pydantic model.

Parameters:

model_type (Type[TargetModel]) – The Pydantic model class to deserialize the parameters into.
parameter (str) – Name or ARN of the parameter
with_decryption (Optional[bool]) – whether decrpytion is applied to the parameter Defaults to False.
ssm_client (Optional[SSMClient]) – An optional Boto3 SSM client. If None, a new client will be created. Defaults to None.

Returns:

An instance of the specified Pydantic model validated from the parameter value as json

Return type:

ModelTypeDef

saigon.aws.ssm.get_parameter_mapping_as_model(model_type, param_mapping, with_decryption=False, ssm_client=None, **kwargs)

Retrieves a list of parameters from AWS SSM and converts them into a Pydantic model.

Parameters:

model_type (Type[TargetModel]) – The Pydantic model class to deserialize the parameters into.
param_mapping (Dict[str]) – Mapping of parameter names or ARNs to model attributes
with_decryption (Optional[bool]) – whether decrpytion is applied to the parameters. Defaults to False.
ssm_client (Optional[SSMClient]) – An optional Boto3 SSM client. If None, a new client will be created. Defaults to None.
**kwargs – extra model parameters if necessary to successfully construct the
object. (model)

Returns:

An instance of the specified Pydantic model populated with the parameters values

Return type:

ModelTypeDef

rest

class saigon.aws.rest.AwsSIGv4RestClient(api_base_url, cognito_config)

Bases: RestClient

A REST client that integrates with AWS Cognito for authentication and SigV4 signing.

This client extends RestClientBase to provide user login capabilities via Cognito and automatically sign all outgoing requests with AWS SigV4, using temporary IAM credentials obtained from Cognito.

class saigon.aws.rest.SIGv4RequestAuthorizer(cognito_client)

Bases: RequestAuthorizer

authorize(request)

Implementation that signs an AWSRequest using SigV4 authentication with the current user’s credentials, which must be present or else this operation will fail.

Parameters:: request (Request) – The Request object to be signed.
Returns:: The signed Request object.
Return type:: Request
Raises:: ValueError – If no user is currently logged in.

property current_user: Tuple[str, UUID] | None

Returns the currently logged-in username and their UUID identity.

Returns:

A tuple containing the username and: their UUID if a user is logged in, otherwise None.

Return type:

Optional[Tuple[str, uuid.UUID]]

login(username, password)

Logs in a user via Cognito and retrieves IAM credentials.

This method performs the user authentication against Cognito, extracts the user’s UUID from the ID token, obtains temporary AWS IAM credentials, and stores them internally for subsequent authenticated requests.

Parameters:

username (str) – The username for login.
password (str) – The password for login.

Returns:

A tuple containing the user’s UUID: and the AWS IAM credentials obtained from Cognito.

Return type:

Tuple[uuid.UUID, SigV4Credentials]

switch_user(username)

Switches the active user for subsequent requests.

The user must have previously logged in using the login method. This updates the internal state to use the credentials of the specified user.

Parameters:: username (str) – The username to switch to.
Returns:: The UUID of the newly active user.
Return type:: uuid.UUID
Raises:: KeyError – If the provided username has not logged in previously.