actingweb code

Subpackages

• actingweb.db.dynamodb package
• actingweb.handlers package

Submodules

actingweb.actor module

class actingweb.actor.Actor(actor_id: str | None = None, config: Any | None = None)

Bases: object

callback_subscription(peerid=None, sub_obj=None, sub=None, diff=None, blob=None)

async callback_subscription_async(peerid=None, sub_obj=None, sub=None, diff=None, blob=None)

Async version - wraps sync method in asyncio.to_thread to prevent blocking.

create(url: str, creator: str, passphrase: str, actor_id: str | None = None, delete: bool = False, trustee_root: str | None = None, hooks: Any = None) → bool

“Creates a new actor and persists it.

If delete is True, any existing actors with same creator value will be deleted. If it is False, the one with the correct passphrase will be chosen (if any)

create_reciprocal_trust(url, secret=None, desc='', relationship='', trust_type='')

Creates a new reciprocal trust relationship locally and by requesting a relationship from a peer actor.

Parameters:

• relationship – The trust type/permission level to request (friend, admin, etc.)

• trust_type – Expected peer mini-app type for validation (optional)

async create_reciprocal_trust_async(url, secret=None, desc='', relationship='', trust_type='')

Async version of create_reciprocal_trust - prevents blocking on peer HTTP calls.

Creates a new reciprocal trust relationship locally and by requesting a relationship from a peer actor.

create_remote_subscription(peerid=None, target=None, subtarget=None, resource=None, granularity=None)

Creates a new subscription at peerid.

async create_remote_subscription_async(peerid=None, target=None, subtarget=None, resource=None, granularity=None)

Async version - wraps sync method in asyncio.to_thread to prevent blocking.

create_subscription(peerid=None, target=None, subtarget=None, resource=None, granularity=None, subid=None, callback=False)

create_verified_trust(baseuri='', peerid=None, approved=False, secret=None, verification_token=None, trust_type=None, peer_approved=None, relationship=None, desc='')

Creates a new trust when requested and call backs to initiating actor to verify relationship.

Parameters:

• trust_type – The peer’s ActingWeb mini-application type URI

• relationship – The trust type/permission level (friend, admin, etc.)

async create_verified_trust_async(baseuri='', peerid=None, approved=False, secret=None, verification_token=None, trust_type=None, peer_approved=None, relationship=None, desc='')

Async version - wraps sync method in asyncio.to_thread to prevent blocking.

delete() → None

Deletes an actor and cleans up all relevant stored data.

The actor row is removed last, so the actor keeps resolving for the duration of the wipe. Existence checks therefore fail open here, and that is why the tombstone is written first: it is readable throughout the wipe, whereas get() cannot distinguish mid-deletion from live. See actingweb.deletion.

delete_peer_trustee(shorttype=None, peerid=None)

delete_properties()

Deletes all properties.

delete_property(name)

Deletes a property name. (DEPRECATED, use actor’s property store!)

delete_reciprocal_trust(peerid=None, delete_peer=False)

Deletes a trust relationship and requests deletion of peer’s relationship as well.

async delete_reciprocal_trust_async(peerid=None, delete_peer=False)

Async version - wraps sync method in asyncio.to_thread to prevent blocking.

delete_remote_subscription(peerid=None, subid=None)

async delete_remote_subscription_async(peerid=None, subid=None)

Async version - wraps sync method in asyncio.to_thread to prevent blocking.

delete_subscription(peerid=None, subid=None, callback=False)

Deletes a specified subscription

get(actor_id: str | None = None) → dict[str, Any] | None

Retrieves an actor from storage or initialises if it does not exist

get_from_creator(creator: str | None = None) → bool

Initialise an actor by matching on creator/email.

Returns True if an actor could be loaded, otherwise False. When multiple actors share the same creator (possible when unique_creator is disabled), the first deterministic match will be selected in order to provide stable behaviour for login flows that do not specify an explicit actor ID.

get_from_property(name: str = 'oauthId', value: str | None = None) → None

Initialise an actor by matching on a stored property.

Use with caution as the property’s value de-facto becomes a security token. If multiple properties are found with the same value, no actor will be initialised. Also note that this is a costly operation as all properties of this type will be retrieved and proceessed.

get_peer_info(url: str, max_retries: int = 3, retry_delay: float = 0.5) → dict[str, Any]

Contacts another actor over http/s to retrieve meta information.

Includes retry logic for transient network failures with exponential backoff.

Note: This sync method blocks the event loop. In FastAPI/uvicorn contexts, use AsyncTrustHandler which calls create_reciprocal_trust_async() instead.

Parameters:

• url – Root URI of a remote actor

• max_retries – Maximum number of retry attempts (default: 3)

• retry_delay – Initial delay between retries in seconds (default: 0.5)

Return type:

dict

Returns:

The json response from the /meta path in the data element and last_response_code/last_response_message set to the results of the https request

Example:

{
    "last_response_code": 200,
    "last_response_message": "OK",
    "data": {}
}

async get_peer_info_async(url: str) → dict[str, Any]

Async version of get_peer_info using httpx.

Contacts another actor over HTTP/S to retrieve meta information without blocking.

Parameters:

url – Root URI of a remote actor

Returns:

Dict with last_response_code, last_response_message, and data

get_peer_trustee(shorttype=None, peerid=None)

Get a peer, either existing or create it as trustee

Will retrieve an existing peer or create a new and establish trust. If no trust exists, a new trust will be established. Use either peerid to target a specific known peer, or shorttype to allow creation of a new peer if none exists

get_properties()

Retrieves properties from db and returns a dict.

get_property(name)

Retrieves a property object named name. (DEPRECATED, use actor’s property store!)

get_subscription(peerid=None, subid=None, callback=False)

Retrieves a single subscription identified by peerid and subid.

get_subscription_obj(peerid=None, subid=None, callback=False)

Retrieves a single subscription identified by peerid and subid.

get_subscriptions(peerid: str | None = None, target: str | None = None, subtarget: str | None = None, resource: str | None = None, callback: bool | None = None) → list[dict[str, Any]] | None

Retrieves subscriptions from db.

Parameters:

• peerid – Filter by peer ID (None = all peers)

• target – Filter by target (None = all targets)

• subtarget – Filter by subtarget (None = all subtargets)

• resource – Filter by resource (None = all resources)

• callback – Filter by callback flag (None = all, False = inbound, True = outbound)

Returns:

List of subscription dictionaries, or None if actor has no ID

get_trust_relationship(peerid=None)

get_trust_relationships(relationship='', peerid='', trust_type='')

Retrieves all trust relationships or filtered.

is_subscription_suspended(target: str, subtarget: str | None = None) → bool

Check if diff registration is suspended for a target/subtarget.

Parameters:

• target – Target resource (e.g., “properties”)

• subtarget – Optional subtarget (e.g., property name)

Returns:

True if suspended, False otherwise

modify(creator: str | None = None) → bool

modify_trust_and_notify(relationship=None, peerid=None, baseuri='', secret='', desc='', approved=None, verified=None, verification_token=None, peer_approved=None, client_name=None, client_version=None, client_platform=None, oauth_client_id=None, last_accessed=None, last_connected_via=None)

Changes a trust relationship and noties the peer if approval is changed.

async modify_trust_and_notify_async(relationship=None, peerid=None, baseuri='', secret='', desc='', approved=None, verified=None, verification_token=None, peer_approved=None, client_name=None, client_version=None, client_platform=None, oauth_client_id=None, last_accessed=None, last_connected_via=None)

Async version of modify_trust_and_notify - prevents blocking on peer notification.

Changes a trust relationship and notifies the peer if approval is changed. Database operations remain synchronous, but peer HTTP notification is async.

register_diffs(target=None, subtarget=None, resource=None, blob=None)

Registers a blob diff against all subscriptions with the correct target, subtarget, and resource.

If resource is set, the blob is expected to be the FULL resource object, not a diff.

Note: Skips registration if the target/subtarget is currently suspended. Use suspend_subscriptions() and resume_subscriptions() to manage suspension.

resume_subscriptions(target: str, subtarget: str | None = None) → int

Resume diff registration and send resync callbacks.

Sends a resync callback to ALL subscriptions on this target/subtarget, telling them to do a full GET to re-sync their state.

Parameters:

• target – Target resource (e.g., “properties”)

• subtarget – Optional subtarget (e.g., property name)

Returns:

The number of resync callbacks sent

set_property(name, value)

Sets an actor’s property name to value. (DEPRECATED, use actor’s property store!)

suspend_subscriptions(target: str, subtarget: str | None = None) → bool

Suspend diff registration for a target/subtarget.

While suspended, property changes will NOT register diffs or trigger callbacks. Call resume_subscriptions() to lift suspension and send resync callbacks.

Parameters:

• target – Target resource (e.g., “properties”)

• subtarget – Optional subtarget (e.g., property name)

Returns:

True if newly suspended, False if already suspended

exception actingweb.actor.ActorError

Bases: Exception

Base exception class for Actor-related errors.

exception actingweb.actor.ActorNotFoundError

Bases: ActorError

Raised when an actor cannot be found.

class actingweb.actor.Actors(config=None)

Bases: object

Handles all actors

fetch()

class actingweb.actor.DummyPropertyClass(v: Any = None)

Bases: object

Only used to deprecate get_property() in 2.4.4

exception actingweb.actor.InvalidActorDataError

Bases: ActorError

Raised when actor data is invalid or corrupted.

exception actingweb.actor.PeerCommunicationError

Bases: ActorError

Raised when communication with peer actors fails.

exception actingweb.actor.TrustRelationshipError

Bases: ActorError

Raised when trust relationship operations fail.

actingweb.attribute module

class actingweb.attribute.Attributes(actor_id: str | None = None, bucket: str | None = None, config: Any | None = None)

Bases: object

Attributes is the main entity keeping an attribute.

It needs to be initalized at object creation time.

conditional_update_attr(name: str | None = None, old_data: Any | None = None, new_data: Any | None = None, timestamp: Any | None = None) → bool

Conditionally update an attribute only if current data matches old_data.

This provides atomic compare-and-swap functionality for race-free updates.

Parameters:

• name – Attribute name

• old_data – Expected current data value (for comparison)

• new_data – New data to set if current matches old_data

• timestamp – Optional timestamp

Returns:

True if update succeeded (current matched old_data), False otherwise

delete_attr(name: str | None = None) → bool

delete_attr_conditional(name: str | None = None) → bool

Atomically delete a single attribute, returning True only if this call removed an existing value.

This is the race-free counterpart to delete_attr(): concurrent callers racing on the same attribute see exactly one True. Use it to enforce single-use consume semantics (e.g. mobile-ticket redemption), where the caller that wins the delete is the only one allowed to act on the value.

delete_bucket() → bool

Deletes the attribute bucket in the database

get_attr(name: str | None = None) → dict[str, Any] | None

Retrieves a single attribute.

A fully-loaded bucket is AUTHORITATIVE: once get_bucket() has returned a dict, a name absent from it is absent from storage, and answering None here costs no query. Before, every absent name cost one point read per instance.

The early return also stops this method polluting a loaded bucket. The miss path below caches self.data[name] = None, and get_bucket() returns self.data BY IDENTITY — so a loaded bucket used to grow keys that have no stored row, and a caller iterating the “bucket” saw names that do not exist.

“Absent” stays distinguishable from “present with a null value”: a stored row holding null reads back as the truthy dict {"data": None, "timestamp": ...}, while absence is None.

get_bucket() → dict[str, Any] | None

Retrieves the attribute bucket from the database.

Tracks full-bucket loads with a flag rather than data emptiness: a set_attr()/get_attr() may have cached individual entries, and treating a partially-cached dict as “loaded” would silently return an incomplete bucket.

_bucket_loaded is set only when the backend actually RETURNED A DICT, and it means “loaded, and the backend answered”. That distinction is load-bearing now that get_attr() treats the flag as authoritative about ABSENCE: a bucket that could not be read must not become “the bucket has no such attribute”, permanently, for the life of this instance.

Both backends return {} for an empty bucket and reserve None for a caught fault (PostgreSQL joined DynamoDB in 3.14.4; before that it returned None for empty too). So an empty bucket is authoritative on both: get_attr(name) after get_bucket() == {} answers None without a backend read for the life of this instance. Every Attributes instance in the library is request-local except InternalStore._db, which loads once via _ensure_loaded on both backends regardless and does not use this path.

Note that on DynamoDB most real faults do not arrive as None at all: DbAttribute.get_bucket() wraps only the Query CONSTRUCTION, and PynamoDB fires the request lazily during iteration, so a throttle mid-page raises straight through this method with the flag still unset — already the safe outcome.

set_attr(name: str | None = None, data: Any | None = None, timestamp: Any | None = None, ttl_seconds: int | None = None) → bool

Sets new data for this attribute.

Parameters:

• name – Attribute name

• data – Data to store (JSON-serializable)

• timestamp – Optional timestamp

• ttl_seconds – Optional TTL in seconds. If provided, DynamoDB will automatically delete this item after expiry.

class actingweb.attribute.Buckets(actor_id: str | None = None, config: Any | None = None)

Bases: object

Handles all attribute buckets of a specific actor_id

Access the attributes in .props as a dictionary

delete() → bool

fetch() → dict[str, dict[str, dict[str, Any]]] | bool

fetch_timestamps() → dict[str, Any] | bool

class actingweb.attribute.InternalStore(actor_id: str | None = None, config: Any | None = None, bucket: str | None = None)

Bases: object

Access to internal attributes using .prop notation

actingweb.auth module

class actingweb.auth.Auth(actor_id, auth_type='basic', config=None)

Bases: object

The auth class handles authentication and authorisation for the various schemes supported.

The check_authentication() function checks the various authentication schemes against the path and does proper authentication. There are two types supported: basic (using creator credentials) and token (received when trust is created or OAuth2). The check_authorisation() function validates the authenticated user against the config.py access list. check_token_auth() can be called from outside the class to do a simple peer/bearer token verification.

The response[], acl[], and authn_done variables are useful outside Auth(). authn_done is set when authentication has been done and a final authentication status can be found in response[].

self.response = {

“code”: 403, # Result code (http) “text”: “Forbidden”, # Proposed response text “headers”: [], # Headers to add to response after authentication has been done

}

self.acl = {

“authenticated”: False, # Has authentication been verified and passed? “authorised”: False, # Has authorisation been done and appropriate acls set? “rights”: ‘’, # “a”, “r” (approve or reject) “relationship”: None, # E.g. creator, friend, admin, etc “peerid”: ‘’, # Peerid if there is a relationship “approved”: False, # True if the peer is approved

}

check_authentication(appreq, path)

Checks authentication in appreq, redirecting back to path if oauth is done.

check_authorisation(path='', subpath='', method='', peerid='', approved=True)

Checks if the authenticated user has acl access rights in config.py.

Takes the path, subpath, method, and peerid of the path (if auth user is different from the peer that owns the path, e.g. creator). If approved is False, then the trust relationship does not need to be approved for access

check_token_auth(appreq, via_hint: str | None = None)

Validate bearer tokens and optionally record how the connection occurred.

async check_token_auth_async(appreq, via_hint: str | None = None)

Async version: Validate bearer tokens without blocking the event loop.

actingweb.auth.add_auth_response(appreq=None, auth_obj=None)

Called after authentication to set appropriate HTTP response based on auth result.

actingweb.auth.check_and_verify_auth(appreq=None, actor_id=None, config=None)

Check and verify authentication for non-ActingWeb routes.

This function provides authentication verification for custom routes that don’t go through the standard ActingWeb handler system. It performs authentication checks and is designed for use in custom application routes.

Parameters:

• appreq – Request object in the format used by ActingWeb handlers.

• actor_id (str | None) – Actor ID to verify authentication against.

• config (Config | None) – ActingWeb config object.

Returns:

A dictionary with the following keys:

• authenticated (bool): True if authentication successful.

• actor (Actor | None): Actor object when authenticated, otherwise None.

• auth (Auth): Auth object with authentication details.

• response (dict): Response details: {"code": int, "text": str, "headers": dict}.

• redirect (str | None): Redirect URL if authentication requires redirect.

Return type:

dict

Example

auth_result = check_and_verify_auth(appreq, actor_id, config)
if not auth_result['authenticated']:
    if auth_result['response']['code'] == 302:
        # Redirect for OAuth
        return redirect(auth_result['redirect'])
    # Return error response
    return error_response(
        auth_result['response']['code'], auth_result['response']['text']
    )

# Authentication successful, use auth_result['actor']
actor = auth_result['actor']

async actingweb.auth.check_and_verify_auth_async(appreq=None, actor_id=None, config=None)

Async version: Check and verify authentication for non-ActingWeb routes.

This async function provides authentication verification for custom routes that don’t go through the standard ActingWeb handler system. It uses async HTTP calls to avoid blocking the event loop during OAuth2 token validation.

Use this in async FastAPI endpoints instead of check_and_verify_auth() when the endpoint might receive OAuth2 tokens that need validation against the provider.

Parameters:

• appreq – Request object in the format used by ActingWeb handlers.

• actor_id (str | None) – Actor ID to verify authentication against.

• config (Config | None) – ActingWeb config object.

Returns:

A dictionary with the following keys:

• authenticated (bool): True if authentication successful.

• actor (Actor | None): Actor object when authenticated, otherwise None.

• auth (Auth): Auth object with authentication details.

• response (dict): Response details: {"code": int, "text": str, "headers": dict}.

• redirect (str | None): Redirect URL if authentication requires redirect.

Return type:

dict

Example

auth_result = await check_and_verify_auth_async(appreq, actor_id, config)
if not auth_result['authenticated']:
    if auth_result['response']['code'] == 302:
        return RedirectResponse(auth_result['redirect'])
    return JSONResponse(
        status_code=auth_result['response']['code'],
        content={"error": auth_result['response']['text']}
    )

# Authentication successful, use auth_result['actor']
actor = auth_result['actor']

actingweb.aw_proxy module

class actingweb.aw_proxy.AwProxy(trust_target: Any = None, peer_target: dict[str, Any] | None = None, config: Any = None, timeout: int | float | tuple[int | float, int | float] | None = None)

Bases: object

Proxy to other trust peers to execute RPC style calls.

Initialise with either trust_target to target a specific existing trust or use peer_target for simplicity to use the trust established with the peer.

Parameters:

• trust_target – Trust object for the target peer

• peer_target – Simplified peer target dict

• config – Configuration object

• timeout – HTTP timeout in seconds. Either a single value (used for both connect and read timeouts) or a tuple (connect_timeout, read_timeout). Default: (5, 20) = 5s connect, 20s read timeout.

Provides both sync methods (using requests) and async methods (using httpx) for peer communication:

• Sync: get_resource(), create_resource(), change_resource(), delete_resource()

• Async: get_resource_async(), create_resource_async(), change_resource_async(), delete_resource_async()

Use async methods in FastAPI routes for non-blocking I/O.

change_resource(path=None, params=None)

async change_resource_async(path: str | None = None, params: dict[str, Any] | None = None) → dict[str, Any] | None

Async version of change_resource (PUT) using httpx.

Parameters:

• path – The resource path on the peer actor

• params – Data to send as JSON body

Returns:

The JSON response from the peer, or None if the request failed.

create_resource(path=None, params=None)

async create_resource_async(path: str | None = None, params: dict[str, Any] | None = None) → dict[str, Any] | None

Async version of create_resource (POST) using httpx.

Parameters:

• path – The resource path on the peer actor

• params – Data to send as JSON body

Returns:

The JSON response from the peer, or None if the request failed.

delete_resource(path=None)

async delete_resource_async(path: str | None = None) → dict[str, Any] | None

Async version of delete_resource (DELETE) using httpx.

Parameters:

path – The resource path on the peer actor

Returns:

The JSON response from the peer, or None if the request failed.

get_resource(path=None, params=None)

async get_resource_async(path: str | None = None, params: dict[str, Any] | None = None) → dict[str, Any] | None

Async version of get_resource using httpx.

Use this method in async contexts (e.g., FastAPI routes) for non-blocking HTTP calls to peer actors.

Parameters:

• path – The resource path on the peer actor (e.g., “trust/friend/permissions”)

• params – Optional query parameters

Returns:

The JSON response from the peer, or None if the request failed.

actingweb.aw_web_request module

class actingweb.aw_web_request.AWRequest(url: str | None = None, params: dict[str, Any] | None = None, body: str | None = None, headers: dict[str, str] | None = None, cookies: dict[str, str] | None = None)

Bases: object

arguments() → list[str]

get(var: str = '') → str

get_header(header: str = '') → str

class actingweb.aw_web_request.AWResponse

Bases: object

set_cookie(name: str, value: str, max_age: int = 1209600, path: str = '/', secure: bool = True, httponly: bool = False, samesite: str = 'Lax') → None

set_redirect(url: str) → None

set_status(code: int = 200, message: str = 'Ok') → bool | None

write(body: str | None = None, encode: bool = False) → bool | None

class actingweb.aw_web_request.AWWebObj(url: str | None = None, params: dict[str, Any] | None = None, body: str | None = None, headers: dict[str, str] | None = None, cookies: dict[str, str] | None = None)

Bases: object

actingweb.config module

class actingweb.config.Config(**kwargs: Any)

Bases: object

DbActor: ModuleType

DbAttribute: ModuleType

DbPeerTrustee: ModuleType

DbProperty: ModuleType

DbSubscription: ModuleType

DbSubscriptionDiff: ModuleType

DbSubscriptionSuspension: ModuleType

DbTrust: ModuleType

static new_token(length: int = 40) → str

static new_uuid(seed: str) → str

update_supported_options() → None

Update aw_supported based on enabled features.

This method is called when features are dynamically enabled after Config initialization (e.g., via ActingWebApp builder methods).

actingweb.peertrustee module

class actingweb.peertrustee.PeerTrustee(actor_id=None, peerid=None, short_type=None, peer_type=None, config=None)

Bases: object

create(baseuri=None, passphrase=None)

delete()

get() → dict[str, Any] | bool | None

Retrieve peer trustee from database.

Returns:

Peer trustee data if exactly one match found False: When multiple peer trustees of same type exist (ambiguous lookup) None: When no peer trustee found or required parameters missing

Return type:

dict[str, Any]

actingweb.property module

class actingweb.property.Properties(actor_id: str | None = None, config: Any | None = None)

Bases: object

Handles all properties of a specific actor_id

Access the properties in .props as a dictionary

delete() → bool

fetch() → dict[str, Any] | bool

class actingweb.property.Property(actor_id: str | None = None, name: str | None = None, value: Any | None = None, config: Any | None = None)

Bases: object

property is the main entity keeping a property.

It needs to be initalised at object creation time.

delete() → bool | None

Deletes the property in the database

get() → Any

Retrieves the property from the database

get_actor_id() → str | None

set(value: Any) → bool

Sets a new value for this property

class actingweb.property.PropertyListStore(actor_id: str | None = None, config: Any | None = None)

Bases: object

Explicit interface for managing list properties.

Used when the application knows it’s working with list data.

exists(name: str) → bool

Check if a list property exists without creating it.

list_all() → list[str]

List all existing list property names.

list_all_with_rows() → tuple[list[str], dict[str, str]]

List all existing list property names, alongside the raw rows the names were derived from.

list_all() already pays for fetch_all_including_lists() – the actor’s WHOLE partition, item rows included – and discards it. This returns that dump too, so a caller who needs both the names and each list’s contents can prime every list from rows already in hand (ListProperty.prime_from_rows() / to_list_from_rows()) instead of paying a second whole-list Query per list.

The rows are a point-in-time snapshot, stale the moment a mutation lands, and OPAQUE: feed them to prime_from_rows() / to_list_from_rows() and never inspect or parse a row name – the encoding (list:{name}-meta, list:{name}-{index} or list:{name}-#{rank}) is a storage detail the next major version’s key-prefix scheme will change.

If you only need ONE namespace of lists, see list_prefix_with_rows(), which reads just that namespace. Note the cost contrast carefully, because it does not go the way the names suggest: on a measured account this dump was 1,361.0 RCU over 11 queries, and the five scoped reads that together cover the same lists were 1,363.5 RCU over 15. Replacing one dump with several scoped reads is marginally WORSE. The scoped method pays when you want one namespace and were dumping the partition to get it – and the latency win from issuing several of them is the CALLER’s, from issuing them concurrently.

On error this returns ([], {}) rather than raising. That is kept for compatibility and is the OPPOSITE of list_prefix_with_rows(); the asymmetry is deliberate and its reasoning is on that method.

list_prefix_with_rows(prefix: str) → tuple[list[str], dict[str, str]]

List properties whose name begins with prefix, and their rows.

The scoped counterpart of list_all_with_rows(): ONE namespace of the actor’s lists instead of the whole partition. Both halves of the return are scoped – names holds only the matching lists, and rows only their rows – so the pair stays internally consistent. That is worth reading twice, because it is where a migration from list_all_with_rows() goes wrong SILENTLY: code that swaps the method and keeps iterating names simply stops seeing every list outside the prefix, with nothing raised.

prefix is a PREFIX, not a namespace. It matches every list whose name begins with it – including a list named exactly prefix, and siblings such as {prefix}-old. If you mean a namespace, pass the delimiter: “memory_”, not “memory”.

Cost, honestly: this is not universally cheaper than the dump. Five scoped reads covering the same lists measured 1,363.5 RCU / 15 queries against the dump’s 1,361.0 / 11 – summing all of them is marginally worse. It pays when you want ONE namespace, or when you can issue several concurrently: the library is synchronous and spends one query per call, so any latency win is the caller’s to take.

Reads are EVENTUALLY consistent, matching what fetch_all_including_lists() already does on DynamoDB (PynamoDB’s Model.query() default), and halving the read capacity – which is what decides whether scoping pays at all. Do not call this to read back a write you just made.

There is deliberately no names-only list_prefix() sibling. A keys-only projection saves no DynamoDB read capacity (a projection still pays for the whole item; measured 1,361.0 either way), so it would break the list_all/list_all_with_rows pairing for nothing.

Rows are OPAQUE and a point-in-time snapshot, exactly as for list_all_with_rows(). Different lists in one result may reflect different instants; there is no snapshot isolation across them.

Parameters:

prefix – The list-name prefix, without the storage list: prefix. Pass “memory_” to reach lists memory_a, memory_b, …

Returns:

(names, rows). Both empty when nothing matches – which for a scoped read is a common, ordinary answer.

names is derived from -meta rows, exactly as in list_all_with_rows(). A damaged list whose meta row was lost therefore contributes rows attributed to no name, rather than having those rows dropped – matching the sibling method, and leaving recoverable data visible. Use rows_for() when you need only the rows belonging to a known set of names.

Raises:

• ValueError – If prefix is empty. An empty prefix would be the whole-partition dump under a name promising the opposite; call list_all_with_rows() when that is what you want.

• DbError – On a backend fault. UNLIKE list_all_with_rows(), which swallows to ([], {}). The asymmetry is deliberate: for a scoped read “nothing here” is the common answer, so swallowing would render a throttled query as “you have no memories” and the caller could not tell.

class actingweb.property.PropertyStore(actor_id: str | None = None, config: Any | None = None)

Bases: object

get_all() → dict[str, Any]

Fetch all properties from the database and return as dictionary.

actingweb.property.rows_for(names: list[str], rows: dict[str, str]) → dict[str, str]

The subset of rows attributable to the given list names.

For narrowing a (names, rows) pair from list_all_with_rows() or list_prefix_with_rows() after pruning names – the permission filter in interface/authenticated_views.py is the caller. It lives here, in the module that owns the row encoding, because interface/ must never parse a row name: list_all_with_rows()’s own docstring declares the rows OPAQUE to callers.

A bare name.startswith(f"list:{name}-") is WRONG and is the reason this exists. For list foo it also claims every row of a sibling named foo-old and foo-5. Used to prune, that silently strips a PERMITTED sibling’s item rows while keeping its -meta row, and to_list_from_rows() then returns [] – a permitted list reported as empty, with nothing raised. So attribution uses the same two shape checks every reader in property_list.py uses: _V1_INDEX_RE for a v1 item row’s -{digits} suffix and _v2_is_rank() for a v2 row’s -#{rank} suffix, plus the exact -meta name.

Rows that belong to no name in names are dropped, including rows of a list that simply was not asked for. Order is not meaningful.

actingweb.subscription module

class actingweb.subscription.Subscription(actor_id=None, peerid=None, subid=None, callback=False, config=None)

Bases: object

Base class with core subscription methods (storage-related)

add_diff(blob=None)

Add a new diff for this subscription

clear_diff(seqnr)

Clears one specific diff

clear_diffs(seqnr=0)

Clear all diffs up to and including a seqnr

create(target: str | None = None, subtarget: str | None = None, resource: str | None = None, granularity: str | None = None, seqnr: int = 0) → bool

Create new subscription and push it to db

decrease_seq()

Rollback sequence number by 1 (used when diff creation fails after seq increment)

delete()

Delete a subscription in storage

get() → dict[str, Any]

Retrieve subscription from db given pre-initialized variables

get_diff(seqnr=0)

Get one specific diff

get_diffs()

Get all the diffs available for this subscription ordered by the timestamp, oldest first

increase_seq()

class actingweb.subscription.Subscriptions(actor_id=None, config=None)

Bases: object

Handles all subscriptions of a specific actor_id

Access the indvidual subscriptions in .dbsubscriptions and the subscription data in .subscriptions as a dictionary

delete()

fetch()

actingweb.trust module

class actingweb.trust.Trust(actor_id: str | None = None, peerid: str | None = None, token: str | None = None, config: Any | None = None)

Bases: object

create(baseuri: str = '', peer_type: str = '', relationship: str = '', secret: str = '', approved: bool = False, verified: bool = False, verification_token: str = '', desc: str = '', peer_approved: bool = False, peer_identifier: str | None = None, established_via: str | None = None, created_at: str | None = None, last_accessed: str | None = None, last_connected_via: str | None = None) → bool

Create a new trust relationship

delete() → bool

Delete the trust relationship

get() → dict[str, Any] | None

Retrieve a trust relationship with either peerid or token

modify(baseuri: str | None = None, secret: str | None = None, desc: str | None = None, approved: bool | None = None, verified: bool | None = None, verification_token: str | None = None, peer_approved: bool | None = None, peer_identifier: str | None = None, established_via: str | None = None, created_at: str | None = None, last_accessed: str | None = None, last_connected_via: str | None = None, client_name: str | None = None, client_version: str | None = None, client_platform: str | None = None, oauth_client_id: str | None = None, aw_supported: str | None = None, aw_version: str | None = None, capabilities_fetched_at: str | None = None) → bool

class actingweb.trust.Trusts(actor_id: str | None = None, config: Any | None = None)

Bases: object

Handles all trusts of a specific actor_id

Access the indvidual trusts in .dbtrusts and the trust data in .trusts as a dictionary

delete() → bool

fetch() → list[dict[str, Any]] | list[Any] | None

actingweb.trust.canonical_connection_method(method: str | None) → str | None

Normalize connection hints to canonical channels.

Parameters:

method – Connection method hint (e.g., “oauth2:interactive”, “trust”, “mcp”)

Returns:

Canonical connection method or original value if not recognized

Examples

• “oauth2:interactive” -> “oauth”

• “OAUTH” -> “oauth”

• “trust” -> “trust”

• “subscription” -> “subscription”

• “mcp” -> “mcp”

• “unknown_method” -> “unknown_method”

actingweb.runtime_context module

Runtime Context System for ActingWeb.

This module provides a generic system for making request-specific context available to hook functions during request processing. This solves the architectural constraint where hook functions have fixed signatures but need access to request-specific context.

Architecture Problem:

• ActingWeb hook functions have fixed signatures: hook(actor, action_name, data)

• Multiple clients can access the same actor (MCP clients, web users, API clients)

• Each request needs context about the current client/request for proper handling

• Can’t modify hook signatures without breaking framework compatibility

• The actor object handed to hooks is frequently the same cached object reused across many requests (MCP’s per-actor cache is the primary example), so context cannot live as mutable state on the actor without leaking between requests and, under concurrency, between callers.

Solution:

• Store context in a module-level contextvars.ContextVar, keyed by actor id, rather than as an attribute on the actor object.

• ContextVar is task-local under asyncio and thread-local under a plain WSGI worker thread, so context set for one request is invisible to a concurrently running request even when both share the same actor object.

• Provide type-safe access methods with clear documentation

• Support multiple context types (MCP, OAuth2, web sessions, etc.)

• Framework integrations (Flask, FastAPI) clear the ContextVar at the end of each request, via the module-level clear_all_context(), so nothing survives onto a reused worker thread. clear_all_context() is framework/test plumbing – application hooks should use RuntimeContext(actor).clear_context() if they need to clear a specific actor’s context mid-request.

Usage Example:

# During request authentication:
runtime_context = RuntimeContext(actor)
runtime_context.set_mcp_context(
    client_id="mcp_abc123",
    trust_relationship=trust_obj,
    peer_id="oauth2_client:user@example.com:mcp_abc123"
)

# In hook functions:
def handle_search(actor, action_name, data):
    runtime_context = RuntimeContext(actor)
    mcp_context = runtime_context.get_mcp_context()
    if mcp_context:
        client_name = mcp_context.trust_relationship.client_name
        # Customize behavior based on client type

Concurrency notes:

• A request dispatched onto a fresh asyncio task automatically inherits the context that was active when the task was created (a snapshot, not a live link — the child cannot mutate the parent’s view and vice versa).

• A request dispatched onto a new thread (e.g. ThreadPoolExecutor) does not automatically inherit the caller’s context; the submitting code must capture contextvars.copy_context() and run the callable via ctx.run(...) for context to cross the hop. See actingweb/interface/hooks.py for the one place ActingWeb does this.

class actingweb.runtime_context.MCPContext(client_id: str, trust_relationship: Any, peer_id: str, token_data: dict[str, Any] | None = None, transport_session_id: str | None = None, client_info: dict[str, Any] | None = None)

Bases: object

Runtime context for MCP (Model Context Protocol) requests.

Contains information about the current MCP client making the request, allowing tools to customize behavior based on client capabilities.

client_id: str

client_info: dict[str, Any] | None = None

peer_id: str

token_data: dict[str, Any] | None = None

transport_session_id: str | None = None

trust_relationship: Any

class actingweb.runtime_context.OAuth2Context(client_id: str, user_email: str, scopes: list[str], token_data: dict[str, Any] | None = None)

Bases: object

Runtime context for OAuth2 authenticated requests.

Contains information about the current OAuth2 session for web or API access.

client_id: str

scopes: list[str]

token_data: dict[str, Any] | None = None

user_email: str

class actingweb.runtime_context.RuntimeContext(actor: Any)

Bases: object

Generic runtime context manager for ActingWeb actors.

Provides type-safe access to request-specific context, stored in a ContextVar keyed by actor id rather than as state on the actor object. This solves the architectural constraint where hook functions can’t receive additional parameters, without letting context leak across requests that happen to share a cached actor object.

The context is request-scoped (task-local under asyncio, thread-local under a plain worker thread) and is cleared by the framework integration at the end of each request.

clear_context() → None

Clear all runtime context for this actor.

This should be called after request processing is complete to avoid context leaking between requests.

get_custom_context(key: str) → Any

Get custom context data.

Parameters:

key – Context key

Returns:

Context value or None if not found

get_mcp_context() → MCPContext | None

Get MCP context for the current request.

Returns:

MCPContext if this is an MCP request, None otherwise

get_oauth2_context() → OAuth2Context | None

Get OAuth2 context for the current request.

Returns:

OAuth2Context if this is an OAuth2 request, None otherwise

get_request_type() → str | None

Determine the type of the current request.

Returns:

“mcp”, “oauth2”, “web”, or None if no context is set

get_web_context() → WebContext | None

Get web browser context for the current request.

Returns:

WebContext if this is a web request, None otherwise

has_context() → bool

Check if any runtime context is set.

Returns:

True if any context is set for this actor

set_custom_context(key: str, value: Any) → None

Set custom context data.

Parameters:

• key – Context key (avoid ‘mcp’, ‘oauth2’, ‘web’ which are reserved)

• value – Context value

set_mcp_context(client_id: str, trust_relationship: Any, peer_id: str, token_data: dict[str, Any] | None = None, transport_session_id: str | None = None, client_info: dict[str, Any] | None = None) → None

Set MCP context for the current request.

Parameters:

• client_id – OAuth2 client ID of the MCP client

• trust_relationship – Trust database record with client metadata

• peer_id – Normalized peer identifier for permission checking

• token_data – Optional OAuth2 token metadata

• transport_session_id – Optional per-MCP-connection identifier that distinguishes concurrent sessions sharing one OAuth2 credential (see MCPContext.transport_session_id).

• client_info – Optional live clientInfo dict from the active session’s initialize call (see MCPContext.client_info).

set_oauth2_context(client_id: str, user_email: str, scopes: list[str], token_data: dict[str, Any] | None = None) → None

Set OAuth2 context for the current request.

Parameters:

• client_id – OAuth2 client ID

• user_email – Authenticated user email

• scopes – Granted OAuth2 scopes

• token_data – Optional token metadata

set_web_context(session_id: str | None = None, user_agent: str | None = None, ip_address: str | None = None, authenticated_user: str | None = None) → None

Set web browser context for the current request.

Parameters:

• session_id – Session identifier

• user_agent – Browser user agent

• ip_address – Client IP address

• authenticated_user – Authenticated user identifier

class actingweb.runtime_context.WebContext(session_id: str | None = None, user_agent: str | None = None, ip_address: str | None = None, authenticated_user: str | None = None)

Bases: object

Runtime context for web browser requests.

Contains session and authentication information for web UI access.

authenticated_user: str | None = None

ip_address: str | None = None

session_id: str | None = None

user_agent: str | None = None

actingweb.runtime_context.clear_all_context() → None

Clear runtime context for every actor tracked in the calling thread’s/task’s own context.

Framework integrations call this once at the end of a request/response cycle (Flask teardown_request, FastAPI middleware finally) as a blanket reset, since a request handler does not always know in advance which actor id(s) it touched. Test fixtures use it to reset state between tests that share a thread or event loop.

This resets only the calling thread’s/task’s own view of _runtime_context_var – it has no effect on a concurrently running request’s context, which lives in its own copied Context.

actingweb.runtime_context.get_client_info_from_context(actor: Any) → dict[str, str] | None

Helper function to extract client information from runtime context.

This provides a unified way to get client details regardless of context type.

Parameters:

actor – Actor object with potential runtime context

Returns:

Dict with ‘name’, ‘version’, ‘platform’ keys, or None if no client info available

actingweb.request_context module

Request context management for logging correlation.

This module provides thread-safe and async-safe storage for request-scoped context information (request ID, actor ID, peer ID) using Python’s contextvars.

The context is automatically isolated per request and propagates correctly through async/await boundaries, making it suitable for both Flask (WSGI) and FastAPI (ASGI) applications.

actingweb.request_context.clear_request_context() → None

Clear all request context values.

This should be called at the end of request handling to prevent context leakage between requests. Framework integrations should call this in finally blocks or response handlers.

Example

>>> set_request_context(actor_id="actor123")
>>> get_actor_id()
'actor123'
>>> clear_request_context()
>>> get_actor_id()  # Returns None

actingweb.request_context.format_context_compact() → str

Format context as a compact string for text-based logging.

Format: [short_request_id:actor_id:short_peer_id] Missing values are represented as “-”

Returns:

actor123:peer456]”

Return type:

Formatted context string like “[a1b2c3d4

Example

>>> set_request_context(
...     request_id="550e8400-e29b-41d4-a716-446655440000",
...     actor_id="actor123",
... )
>>> set_peer_id("urn:actingweb:example.com:peer456")
>>> format_context_compact()
'[55440000:actor123:peer456]'
>>> clear_request_context()
>>> format_context_compact()
'[-:-:-]'

actingweb.request_context.generate_request_id() → str

Generate a new UUID4 request ID.

Returns:

“550e8400-e29b-41d4-a716-446655440000”

Return type:

A string UUID in the format

Example

>>> request_id = generate_request_id()
>>> len(request_id)
36

actingweb.request_context.get_actor_id() → str | None

Get the actor ID for the current context.

Returns:

The actor ID if set, otherwise None

Example

>>> get_actor_id()  # Returns None if not set
>>> set_actor_id("actor123")
>>> get_actor_id()
'actor123'

actingweb.request_context.get_context_dict() → dict[str, Any]

Get all context values as a dictionary.

This is useful for structured logging (JSON) where context should be included as separate fields rather than in a formatted string.

Returns:

request_id, actor_id, peer_id Values are None if not set in current context

Return type:

Dictionary with keys

Example

>>> set_request_context(
...     request_id="550e8400-e29b-41d4-a716-446655440000",
...     actor_id="actor123",
... )
>>> get_context_dict()
{'request_id': '550e8400-e29b-41d4-a716-446655440000', 'actor_id': 'actor123', 'peer_id': None}

actingweb.request_context.get_peer_id() → str | None

Get the peer ID for the current context.

Returns:

The peer ID if set, otherwise None

Example

>>> get_peer_id()  # Returns None if not set
>>> set_peer_id("peer456")
>>> get_peer_id()
'peer456'

actingweb.request_context.get_request_id() → str | None

Get the request ID for the current context.

Returns:

The request ID if set, otherwise None

Example

>>> get_request_id()  # Returns None if not set
>>> set_request_id("550e8400-e29b-41d4-a716-446655440000")
>>> get_request_id()
'550e8400-e29b-41d4-a716-446655440000'

actingweb.request_context.get_short_peer_id() → str

Get a shortened version of the peer ID (last segment after final colon).

Peer IDs often follow the pattern “urn:actingweb:example.com:actor123”, so this extracts just the last segment for compact logging.

Returns:

Last segment of peer ID after final colon, or “-” if no peer ID is set

Example

>>> set_peer_id("urn:actingweb:example.com:actor123")
>>> get_short_peer_id()
'actor123'
>>> set_peer_id("simple_peer")
>>> get_short_peer_id()
'simple_peer'
>>> set_peer_id(None)
>>> get_short_peer_id()
'-'

actingweb.request_context.get_short_request_id() → str

Get a shortened version of the request ID (last 8 characters).

This is useful for compact log formats where the full UUID is too verbose. 8 characters provides sufficient uniqueness for correlation within a reasonable time window.

Returns:

Last 8 characters of the request ID, or “-” if no request ID is set

Example

>>> set_request_id("550e8400-e29b-41d4-a716-446655440000")
>>> get_short_request_id()
'40000'
>>> set_request_id(None)
>>> get_short_request_id()
'-'

actingweb.request_context.set_actor_id(actor_id: str | None) → None

Set the actor ID for the current context.

Parameters:

actor_id – The actor ID to set, or None to clear

Example

>>> set_actor_id("actor123")
>>> get_actor_id()
'actor123'

actingweb.request_context.set_peer_id(peer_id: str | None) → None

Set the peer ID for the current context.

The peer ID identifies the remote actor or client making the request, and is typically set after authentication completes.

Parameters:

peer_id – The peer ID to set, or None to clear

Example

>>> set_peer_id("peer456")
>>> get_peer_id()
'peer456'

actingweb.request_context.set_request_context(request_id: str | None = None, actor_id: str | None = None, peer_id: str | None = None, *, generate_id: bool = True) → str

Set all request context values at once.

This is the primary entry point for framework integrations (Flask, FastAPI) to establish request context at the start of request handling.

Parameters:

• request_id – The request ID, or None to generate a new one

• actor_id – The actor ID from the request path

• peer_id – The peer ID (typically set later after authentication)

• generate_id – If True and request_id is None, generate a new UUID

Returns:

The request ID that was set (either provided or generated)

Example

>>> # Framework integration: set context at request start
>>> req_id = set_request_context(
...     request_id=request.headers.get("X-Request-ID"),
...     actor_id=request.path_params.get("actor_id"),
...     generate_id=True,
... )
>>> print(f"Handling request {req_id}")

actingweb.request_context.set_request_id(request_id: str | None) → None

Set the request ID for the current context.

Parameters:

request_id – The request ID to set, or None to clear

Example

>>> set_request_id("550e8400-e29b-41d4-a716-446655440000")
>>> get_request_id()
'550e8400-e29b-41d4-a716-446655440000'

actingweb.logging_config module

Centralized logging configuration for ActingWeb.

This module provides utilities for configuring logging across the ActingWeb framework with sensible defaults for different environments.

actingweb.logging_config.configure_actingweb_logging(level: int = 20, *, db_level: int | None = None, handlers_level: int | None = None, interface_level: int | None = None, oauth_level: int | None = None, mcp_level: int | None = None) → None

Configure ActingWeb logging with sensible defaults.

This function sets up hierarchical logging for different ActingWeb subsystems, allowing fine-grained control over verbosity in different parts of the framework.

Parameters:

• level – Default level for all actingweb loggers (default: INFO)

• db_level – Override for database operations (default: WARNING to reduce noise)

• handlers_level – Override for HTTP handlers (default: uses main level)

• interface_level – Override for interface layer (default: uses main level)

• oauth_level – Override for OAuth2 components (default: uses main level)

• mcp_level – Override for MCP protocol (default: uses main level)

Example

Development setup (verbose):

>>> import logging
>>> from actingweb.logging_config import configure_actingweb_logging
>>> configure_actingweb_logging(logging.DEBUG)

Production setup (quiet DB, normal handlers):

>>> configure_actingweb_logging(
...     level=logging.WARNING,
...     handlers_level=logging.INFO,
...     db_level=logging.ERROR,
... )

Testing setup (only errors):

>>> configure_actingweb_logging(logging.ERROR)

actingweb.logging_config.configure_actingweb_logging_with_context(level: int = 20, *, db_level: int | None = None, handlers_level: int | None = None, interface_level: int | None = None, oauth_level: int | None = None, mcp_level: int | None = None, enable_context: bool = True, structured: bool = False) → None

Configure ActingWeb logging with request context support.

This is a convenience function that combines configure_actingweb_logging() with enable_request_context_filter() and format configuration.

Parameters:

• level – Default level for all actingweb loggers (default: INFO)

• db_level – Override for database operations (default: WARNING)

• handlers_level – Override for HTTP handlers (default: uses main level)

• interface_level – Override for interface layer (default: uses main level)

• oauth_level – Override for OAuth2 components (default: uses main level)

• mcp_level – Override for MCP protocol (default: uses main level)

• enable_context – If True, enable request context injection (default: True)

• structured – If True, use structured context for JSON logging (default: False)

Example

Development with context:

>>> import logging
>>> configure_actingweb_logging_with_context(logging.DEBUG)

Production with context:

>>> configure_actingweb_logging_with_context(
...     level=logging.WARNING,
...     handlers_level=logging.INFO,
...     db_level=logging.ERROR,
... )

Disable context:

>>> configure_actingweb_logging_with_context(
...     level=logging.INFO,
...     enable_context=False,
... )

actingweb.logging_config.configure_development_logging(*, verbose: bool = False) → None

Configure logging for development environments.

Parameters:

verbose – If True, enable DEBUG logging everywhere (default: False)

Example

>>> configure_development_logging()  # INFO level
>>> configure_development_logging(verbose=True)  # DEBUG level

actingweb.logging_config.configure_production_logging(*, http_traffic: bool = True, lifecycle_events: bool = True) → None

Configure logging for production environments with performance focus.

This is an opinionated production configuration that balances observability with performance.

Parameters:

• http_traffic – If True, log HTTP request handling at INFO level

• lifecycle_events – If True, log actor lifecycle events at INFO level

Example

>>> configure_production_logging()
>>> # HTTP traffic and lifecycle events logged, everything else quiet

actingweb.logging_config.configure_testing_logging(*, debug: bool = False) → None

Configure logging for test environments.

By default, only shows errors during tests to keep output clean. Can be overridden with debug=True for troubleshooting.

Parameters:

debug – If True, show all DEBUG logs (default: False)

Example

>>> import os
>>> # Enable debug logging with environment variable
>>> debug_tests = os.getenv("ACTINGWEB_DEBUG") == "1"
>>> configure_testing_logging(debug=debug_tests)

actingweb.logging_config.enable_request_context_filter(*, logger: str | Logger = 'actingweb', structured: bool = False, handler_type: Literal['all', 'stream', 'file'] = 'all') → None

Enable request context injection for ActingWeb loggers.

This function adds RequestContextFilter to existing logging handlers, enabling automatic injection of request context (request ID, actor ID, peer ID) into log records.

After enabling the filter, you should update log formats to include the %(context)s placeholder:

formatter = logging.Formatter(get_context_format()) handler.setFormatter(formatter)

Parameters:

• logger – Logger name or Logger object to add filters to (default: “actingweb”)

• structured – If True, use StructuredContextFilter for JSON logging; if False, use RequestContextFilter for text logging

• handler_type – Which handlers to add filter to: - “all”: Add to all handlers - “stream”: Add only to StreamHandler instances - “file”: Add only to FileHandler instances

Example

Text logging with context:

>>> import logging
>>> from actingweb.logging_config import (
...     configure_actingweb_logging,
...     enable_request_context_filter,
...     get_context_format,
... )
>>> configure_actingweb_logging(logging.INFO)
>>> enable_request_context_filter()
>>> # Update format to include context
>>> for handler in logging.getLogger("actingweb").handlers:
...     formatter = logging.Formatter(get_context_format())
...     handler.setFormatter(formatter)

JSON logging with structured context:

>>> enable_request_context_filter(structured=True)
>>> # Use custom JSON formatter that accesses record.request_id, etc.

actingweb.logging_config.get_context_format(*, include_timestamp: bool = True, include_context: bool = False, include_logger: bool = True, include_level: bool = True) → str

Generate a log format string with optional request context.

This function creates format strings suitable for use with logging.Formatter. When include_context is True, the format includes a %(context)s placeholder that will be populated by RequestContextFilter.

IMPORTANT: If include_context=True, you MUST add RequestContextFilter to your handlers using enable_request_context_filter() or add_context_filter_to_handler(), otherwise formatting will fail with “Formatting field not found in record: ‘context’”.

Parameters:

• include_timestamp – Include timestamp in format (default: True)

• include_context – Include request context placeholder (default: False)

• include_logger – Include logger name (default: True)

• include_level – Include log level (default: True)

Returns:

A format string suitable for logging.Formatter

Example

>>> get_context_format()
'%(asctime)s %(name)s:%(levelname)s: %(message)s'
>>> get_context_format(include_context=True)
'%(asctime)s %(context)s %(name)s:%(levelname)s: %(message)s'
>>> get_context_format(include_timestamp=False, include_context=False)
'%(name)s:%(levelname)s: %(message)s'

actingweb.logging_config.get_performance_critical_loggers() → list[str]

Return list of loggers that should be WARNING+ in production.

These loggers are in hot paths and excessive logging impacts performance. Use this to configure production environments where performance is critical.

Returns:

List of logger names that are performance-sensitive

Example

>>> for logger_name in get_performance_critical_loggers():
...     logging.getLogger(logger_name).setLevel(logging.WARNING)

actingweb.log_filter module

Logging filters for request context injection.

This module provides logging.Filter classes that automatically inject request context (request ID, actor ID, peer ID) into log records, enabling correlation and filtering without modifying existing log statements.

class actingweb.log_filter.RequestContextFilter(name='')

Bases: Filter

Logging filter that injects request context into log records.

This filter adds a ‘context’ attribute to each LogRecord containing formatted context information. The context includes: - Request ID (short form, last 8 chars) - Actor ID (full) - Peer ID (short form, last segment after colon)

The context is formatted as: [req_id:actor_id:peer_id] Missing values are represented as “-”

Usage:

>>> import logging
>>> from actingweb.log_filter import RequestContextFilter
>>> from actingweb.request_context import set_request_context
>>>
>>> # Set up logger with filter
>>> logger = logging.getLogger("myapp")
>>> handler = logging.StreamHandler()
>>> handler.addFilter(RequestContextFilter())
>>> formatter = logging.Formatter(
...     "%(asctime)s %(context)s %(name)s:%(levelname)s: %(message)s"
... )
>>> handler.setFormatter(formatter)
>>> logger.addHandler(handler)
>>>
>>> # Context automatically appears in logs
>>> set_request_context(actor_id="actor123")
>>> logger.info("Processing request")
# Output: 2024-01-15 10:23:45,123 [a1b2c3d4:actor123:-] myapp:INFO: Processing request

None - all state comes from contextvars in request_context module

filter(record: LogRecord) → bool

Add context to the log record.

This method is called by the logging framework for each log record. It adds a ‘context’ attribute containing the formatted context string.

Parameters:

record – The LogRecord to process

Returns:

Always True (record is never filtered out)

class actingweb.log_filter.StructuredContextFilter(name='')

Bases: Filter

Logging filter that injects request context as separate fields.

This filter is designed for structured logging (JSON) where context values should be separate fields rather than a formatted string.

The filter adds these attributes to each LogRecord: - request_id: Full request ID (UUID) - actor_id: Actor ID - peer_id: Peer ID

Values are None if not set in current context.

Usage:

>>> import logging
>>> import json
>>> from actingweb.log_filter import StructuredContextFilter
>>> from actingweb.request_context import set_request_context
>>>
>>> # Set up logger with filter for JSON output
>>> logger = logging.getLogger("myapp")
>>> handler = logging.StreamHandler()
>>> handler.addFilter(StructuredContextFilter())
>>>
>>> # Custom JSON formatter (simplified example)
>>> class JsonFormatter(logging.Formatter):
...     def format(self, record):
...         return json.dumps({
...             "timestamp": record.created,
...             "level": record.levelname,
...             "logger": record.name,
...             "message": record.getMessage(),
...             "request_id": getattr(record, "request_id", None),
...             "actor_id": getattr(record, "actor_id", None),
...             "peer_id": getattr(record, "peer_id", None),
...         })
>>>
>>> handler.setFormatter(JsonFormatter())
>>> logger.addHandler(handler)
>>>
>>> # Context automatically included in JSON
>>> set_request_context(actor_id="actor123")
>>> logger.info("Processing request")
# Output: {"timestamp": 1234567890.123, "level": "INFO", ...}

None - all state comes from contextvars in request_context module

filter(record: LogRecord) → bool

Add context fields to the log record.

This method is called by the logging framework for each log record. It adds separate attributes for each context field.

Parameters:

record – The LogRecord to process

Returns:

Always True (record is never filtered out)

actingweb.log_filter.add_context_filter_to_handler(handler: Handler, *, structured: bool = False) → None

Add a request context filter to a logging handler.

This is a convenience function for adding the appropriate filter type to an existing handler.

Parameters:

• handler – The logging handler to add the filter to

• structured – If True, use StructuredContextFilter for JSON logging; if False, use RequestContextFilter for text logging

Example

>>> import logging
>>> handler = logging.StreamHandler()
>>> add_context_filter_to_handler(handler)
>>> # Handler now has RequestContextFilter attached

actingweb.log_filter.add_context_filter_to_logger(logger: Logger | str, *, structured: bool = False) → None

Add a request context filter to all handlers of a logger.

This is a convenience function for adding filters to an existing logger’s handlers. It’s useful for adding context to loggers that are already configured.

Parameters:

• logger – The logger (or logger name) to add filters to

• structured – If True, use StructuredContextFilter for JSON logging; if False, use RequestContextFilter for text logging

Example

>>> import logging
>>> logging.basicConfig()  # Set up basic logging
>>> add_context_filter_to_logger("actingweb")
>>> # All actingweb loggers now include request context