emailnator.asyncio.builders.helpers package

Submodules

emailnator.asyncio.builders.helpers.metaclass module

Asynchronous singleton metaclass.

This module defines AsyncSingletonMeta, a metaclass that ensures a class has only one instance in asynchronous contexts. It provides thread-safe singleton behavior using an asyncio.Lock and supports asynchronous initialization via the __ainit__ method.

Typical Usage Example:

class MySingleton(metaclass=AsyncSingletonMeta):

async def __ainit__(self):

await asyncio.sleep(1) self.value = 42

async def main():

obj1 = await MySingleton() obj2 = await MySingleton() assert obj1 is obj2

class emailnator.asyncio.builders.helpers.metaclass.AsyncSingletonMeta

Bases: type

Asynchronous metaclass for implementing singleton classes with thread safety.

This metaclass ensures that only one instance of a class exists, even in asynchronous contexts with potential concurrent instantiation. It uses an asyncio.Lock to guarantee that only one coroutine initializes the singleton at a time. If the class defines an asynchronous initializer (__ainit__), it will be awaited during instance creation.

_instances

A mapping of classes to their singleton instances.

Type:

dict

_lock

A global lock used to prevent race conditions during singleton initialization.

Type:

asyncio.Lock

emailnator.asyncio.builders.helpers.xsrf_token_service module

XSRF token management module for EmailNator API.

This module provides the XsrfManager class, which is responsible for handling the lifecycle of XSRF authentication tokens used when making requests to the EmailNator API. It ensures that valid tokens are fetched, decoded, refreshed, and included in the request headers.

The manager is designed for asynchronous environments and uses an asyncio.Lock to guarantee safe concurrent access to token initialization and refresh operations.

Typical Usage Example:

import httpx, asyncio from emailnator.security.xsrf_manager import XsrfManager

async def main():

async with httpx.AsyncClient() as client:

manager = XsrfManager(client) headers = await manager.get_headers() print(headers)

asyncio.run(main())

class emailnator.asyncio.builders.helpers.xsrf_token_service.XsrfManager(client)

Bases: object

Manage XSRF token lifecycle for the EmailNator API.

This class is responsible for securely handling XSRF tokens used in requests to the EmailNator API. It provides methods to fetch, decode, refresh, and ensure that a valid token and corresponding HTTP headers are always available.

_client

The asynchronous HTTP client used for making requests to the EmailNator API.

Type:

httpx.AsyncClient

_token

Cached XSRF token. None if not yet initialized.

Type:

str | None

_headers

Default HTTP headers containing the XSRF token and other required metadata.

Type:

dict[str, str]

_lock

Concurrency lock to ensure thread-safe token initialization and refresh operations.

Type:

asyncio.Lock

async ensure_token()

Ensure that an XSRF authentication token is available.

If no token is currently stored, this method acquires an asynchronous lock to safely initialize it. A raw token is fetched and decoded, after which the authentication headers are set up.

Raises:

RuntimeError – If the raw token could not be retrieved from cookies.

Return type:

None

async get_headers()

Retrieve a copy of the current request headers.

This method ensures that a valid XSRF token is available before returning the headers. The headers are returned as a shallow copy to prevent external modifications from affecting the internal state.

Returns:

A copy of the current request headers.

Return type:

dict[str, str]

async get_token()

Retrieve the current authentication token.

This method ensures that a valid token is available by calling ensure_token(). If no token is present after validation, an error is raised.

Returns:

The current authentication token.

Return type:

str

Raises:

RuntimeError – If the token could not be obtained.

async refresh()

Refresh the XSRF authentication token.

This method acquires an asynchronous lock to ensure that only one refresh operation is performed at a time. It fetches a new raw token, decodes it, and updates the internal authentication headers.

Raises:

RuntimeError – If the raw token could not be fetched or decoded.

Return type:

None

Module contents

class emailnator.asyncio.builders.helpers.AsyncSingletonMeta

Bases: type

Asynchronous metaclass for implementing singleton classes with thread safety.

This metaclass ensures that only one instance of a class exists, even in asynchronous contexts with potential concurrent instantiation. It uses an asyncio.Lock to guarantee that only one coroutine initializes the singleton at a time. If the class defines an asynchronous initializer (__ainit__), it will be awaited during instance creation.

_instances

A mapping of classes to their singleton instances.

Type:

dict

_lock

A global lock used to prevent race conditions during singleton initialization.

Type:

asyncio.Lock

class emailnator.asyncio.builders.helpers.XsrfManager(client)

Bases: object

Manage XSRF token lifecycle for the EmailNator API.

This class is responsible for securely handling XSRF tokens used in requests to the EmailNator API. It provides methods to fetch, decode, refresh, and ensure that a valid token and corresponding HTTP headers are always available.

_client

The asynchronous HTTP client used for making requests to the EmailNator API.

Type:

httpx.AsyncClient

_token

Cached XSRF token. None if not yet initialized.

Type:

str | None

_headers

Default HTTP headers containing the XSRF token and other required metadata.

Type:

dict[str, str]

_lock

Concurrency lock to ensure thread-safe token initialization and refresh operations.

Type:

asyncio.Lock

async ensure_token()

Ensure that an XSRF authentication token is available.

If no token is currently stored, this method acquires an asynchronous lock to safely initialize it. A raw token is fetched and decoded, after which the authentication headers are set up.

Raises:

RuntimeError – If the raw token could not be retrieved from cookies.

Return type:

None

async get_headers()

Retrieve a copy of the current request headers.

This method ensures that a valid XSRF token is available before returning the headers. The headers are returned as a shallow copy to prevent external modifications from affecting the internal state.

Returns:

A copy of the current request headers.

Return type:

dict[str, str]

async get_token()

Retrieve the current authentication token.

This method ensures that a valid token is available by calling ensure_token(). If no token is present after validation, an error is raised.

Returns:

The current authentication token.

Return type:

str

Raises:

RuntimeError – If the token could not be obtained.

async refresh()

Refresh the XSRF authentication token.

This method acquires an asynchronous lock to ensure that only one refresh operation is performed at a time. It fetches a new raw token, decodes it, and updates the internal authentication headers.

Raises:

RuntimeError – If the raw token could not be fetched or decoded.

Return type:

None