chapps.rest.routers.live module

Routes for live access to CHAPPS state

Please note that documentation about API routes is best viewed in the OpenAPI docs generated by FastAPI when the API is launched.

chapps.rest.routers.live.api = <fastapi.routing.APIRouter object>

The live routes API router

Parameters

scope (MutableMapping[str, Any]) –
receive (Callable[[], Awaitable[MutableMapping[str, Any]]]) –
send (Callable[[MutableMapping[str, Any]], Awaitable[None]]) –

Return type

None

chapps.rest.routers.live.load_user_with_quota(item_id, name)

Create a load_user_with_quota function, using the factory.

Parameters

item_id (int) – ID of user to load, or 0 if loading by name
name (Optiona[str]) – exact name of User to load, if item_id is zero

Returns

a tuple of (User, dict(quota=Quota), [remarks,…])

Because the factory is meant to work with any number of associations, the return format is a little clunky for this case but it seems a small price to pay to avoid code duplication.

Also note that FastAPI interprets the docstrings of the coroutines defined in this module, in order to produce the live API documentation. As a result, they have been kept simpler than in some other modules.

chapps.rest.routers.live.load_users_with_quota(item_ids)

Create a username to quota id mapping function

Parameters: item_ids (List[int]) – list of User ids to associate to quotas
Returns: a list of user ORM objects with associated quotas

async chapps.rest.routers.live.get_bulk_quota_remaining(user_ids)[source]

Accepts a list of user ids.

Returns a list of JSON objects with attributes named user_name and quota_avail, representing the remaining number of email transmissions available to that user at the moment the query was executed.

Return type: BulkQuotaResp
Parameters: user_ids (List[int]) –

async chapps.rest.routers.live.get_current_quota_remaining_for_user(user_id=0, name=Body(None))[source]

Accepts the id of the user whose remaining quota should be checked.

Returns the instantaneous number of available send attempts in response

Return type

LiveQuotaResp

Parameters

user_id (int) –
name (Optional[str]) –

async chapps.rest.routers.live.reset_live_quota_for_user(user_id=0, name=Body(None))[source]

Accepts the id of the user whose quota should be reset.

Returns the number of send attempts dropped in response

Return type

LiveQuotaResp

Parameters

user_id (int) –
name (Optional[str]) –

async chapps.rest.routers.live.refresh_quota_policy_for_user(user_id=0, name=Body(None))[source]

Accepts the id of the user whose quota policy should be refreshed.

Returns the new remaining quota after the policy update in response

Return type

LiveQuotaResp

Parameters

user_id (int) –
name (Optional[str]) –

async chapps.rest.routers.live.refresh_config_on_disk(passcode=Body(Ellipsis))[source]

Writes the current effective config to disk.

Requires the CHAPPS password to be provided as the body.

If line transmission security is an issue: an SSL proxy layer will be required. This is true for the entire application.

Return type: TextResp
Parameters: passcode (str) –

async chapps.rest.routers.live.sda_batch_peek(user_ids, domain_ids=[], email_ids=[])[source]

Accepts domain_ids, email_ids and user_ids as body arguments: lists of integer object ids.

Looks at current authorizations for all domain-user combinations.

Returns their cache status as a dict of dicts, keyed as: sda[domain-or-email][user] = SDAStatus

Return type

SourceUserMapResp

Parameters

user_ids (List[int]) –
domain_ids (List[int]) –
email_ids (List[int]) –

async chapps.rest.routers.live.sda_peek(source_name, user_name)[source]

Accepts url-encoded domain name or email address and user name as path arguments.

Returns status of cached SDA for the named user and source, i.e. is this user allowed to transmit email apparently from this domain or email address

Return type

TextResp

Parameters

source_name (str) –
user_name (str) –

async chapps.rest.routers.live.sda_batch_clear(user_ids, domain_ids=[], email_ids=[])[source]

Clears all source - user mappings by iterating through both lists: users and domains+emails.

Return type

TextResp

Parameters

user_ids (List[int]) –
domain_ids (List[int]) –
email_ids (List[int]) –

async chapps.rest.routers.live.sda_clear(source_name, user_name)[source]

Accepts url-encoded domain name or email address and user name of SDA to clear.

Returns the status of the SDA prior to clearing.

Return type

TextResp

Parameters

source_name (str) –
user_name (str) –

async chapps.rest.routers.live.grl_peek_tuple(client_address=Body(Ellipsis), sender=Body(Ellipsis), recipient=Body(Ellipsis))[source]

Accepts client IP address, sender email address and recipient email address as required arguments in the request body.

Returns a float which is the UNIX epoch time of the last time that tuple was encountered by greylisting.

Parameters

client_address (str) –
sender (str) –
recipient (str) –

async chapps.rest.routers.live.grl_list_tally(client_address)[source]

Accepts the client IP address as the path argument.

Returns a list of instance IDs and their timestamps as floats in UNIX epoch time (UTC). In the standard time library, localtime() will convert them to a time struct in local time based on locale, or gmtime() will convert them to a struct in UTC. Then strftime() may be used to format them for a human to read.

Parameters: client_address (str) –

async chapps.rest.routers.live.grl_clear_tally(client_address)[source]

Accepts the client IP address as the path argument.

Returns “deleted” if successful.

Parameters: client_address (str) –

async chapps.rest.routers.live.grl_clear_option_cache(recipient_domain)[source]

Accepts a domain name as the path argument.

Returns ‘deleted’ on success.

Parameters: recipient_domain (str) –

async chapps.rest.routers.live.spf_clear_option_cache(recipient_domain)[source]

Accepts a domain name as the path argument.

Returns ‘deleted’ on success.

Parameters: recipient_domain (str) –