python3-commons

Re-usable code for Python 3 projects.

Installation

uv add python3-commons

Optional Dependencies

Some features require extra dependencies. You can install them individually or all at once:

• api-client: For python3_commons.api_client
• audit: For python3_commons.audit
• authn: For python3_commons.auth
• authz: For python3_commons.permissions
• cache: For python3_commons.cache
• database: For python3_commons.db
• object-storage: For python3_commons.object_storage
• soap-client: For python3_commons.soap_client
• all: Install all optional dependencies

uv add "python3-commons[all]"

Features

Async LRU Cache

An LRU cache decorator for asynchronous functions that handles await and prevents the dogpile effect (mitigating multiple concurrent calls for the same key by waiting for the first one to finish).

frompython3_commons.async_functoolsimportasync_lru_cache@async_lru_cache(maxsize=128)asyncdeffetch_expensive_data(item_id: int):
# This function will only be called once for a given item_id # even if multiple tasks await it simultaneously.returnawaitdb.get(item_id)
# Usageresult=awaitfetch_expensive_data(42)

API Client

A context manager for aiohttp requests with built-in audit logging to S3 and standardized error mapping to Python exceptions.

frompython3_commons.api_clientimportrequestfromaiohttpimportClientSessionasyncwithClientSession() assession:
asyncwithrequest(
session, base_url="https://api.example.com", uri="/data", method="get",
audit_name="my_service_audit"
) asresponse:
data=awaitresponse.json()

SOAP Client

Async SOAP client support for zeep with S3 auditing capabilities.

frompython3_commons.soap_clientimportsoap_clientasyncwithsoap_client("https://example.com/service?wsdl") asclient:
result=awaitclient.service.GetData(id=42)

Database Management

SQLAlchemy async engine and session management with pool tuning, health checks, and dynamic query builders.

frompython3_commons.dbimportAsyncSessionManagerfrompython3_commons.confimportDBSettings# Configurationconfigs= {"default": DBSettings(dsn="postgresql+asyncpg://user:pass@localhost/db")}
manager=AsyncSessionManager(configs)
# Usageasyncwithmanager.get_session_context("default") assession:
result=awaitsession.execute(...)
# Health checkfrompython3_commons.dbimportis_healthyawaitis_healthy(manager.get_engine("default"))

Object Storage

Utilities for async S3 operations using aiobotocore.

frompython3_commonsimportobject_storageimportio# Upload an objectawaitobject_storage.put_object(
bucket_name="my-bucket", path="uploads/file.txt", data=io.BytesIO(b"Hello World"), length=11
)
# Download an objectcontent=awaitobject_storage.get_object("my-bucket", "uploads/file.txt")
# List objectsasyncforobjinobject_storage.list_objects("my-bucket", "uploads/"):
print(obj['Key'])

OIDC Authentication

Client for OpenID Connect authentication, supporting configuration fetching, JWKS, and token acquisition.

frompython3_commons.authimportOIDCClientfrompydanticimportHttpUrlclient=OIDCClient(
authority_url=HttpUrl("https://auth.example.com/realms/myrealm"),
client_id="my-app-client",
client_secret="secret"
)
asyncwithclient:
token_response=awaitclient.fetch_token(username="user", password="password")
print(token_response.access_token)

Valkey Cache

Async caching using Valkey (Redis-compatible) with automatic Msgpack serialization for complex types.

frompython3_commonsimportcache# Store a dictionaryawaitcache.store("user:123", {"name": "Alice", "role": "admin"}, ttl=3600)
# Retrieve ituser_data=awaitcache.get("user:123")
# Set operationsawaitcache.add_set_item("active_users", "user:123")
is_active=awaitcache.has_set_item("active_users", "user:123")

Structured Logging

A JSONFormatter for structured logging, compatible with standard Python logging.

importloggingfrompython3_commons.log.formattersimportJSONFormatterhandler=logging.StreamHandler()
handler.setFormatter(JSONFormatter())
logging.getLogger().addHandler(handler)
logger=logging.getLogger("app")
logger.info("User logged in", extra={"user_id": "abc-123", "ip": "1.2.3.4"})

Serialization

Enhanced JSON and Msgpack serialization for types not handled by default (Decimal, datetime, date, dataclasses, Pydantic models).

frompython3_commons.serializers.msgspecimportserialize_msgpack, deserialize_msgpackfromdecimalimportDecimalfromdatetimeimportdatetimedata= {
"amount": Decimal("150.75"),
"timestamp": datetime.now(),
"tags": {"finance", "internal"}
}
# Serialize to Msgpackbinary=serialize_msgpack(data)
# Deserialize backrestored=deserialize_msgpack(binary)

RBAC Permissions

Database-backed Role-Based Access Control (RBAC) permission checking.

frompython3_commons.permissionsimporthas_user_permissionfromuuidimportUUIDuser_uuid=UUID("...")
allowed=awaithas_user_permission(session, user_uuid, "reports.view")

General Helpers

A collection of useful utility functions:

• to_snake_case(text): Converts strings to snake_case.
• round_decimal(value, places): Rounds Decimal values.
• tries(n): An async retry decorator.
• log_execution_time: An async decorator to log how long a function takes.
• date_from_string / datetime_from_string: Flexible date/time parsing.
• request_to_curl: Converts request parameters to a curl command string.

frompython3_commons.helpersimporttries, log_execution_time@tries(3)@log_execution_timeasyncdefflaky_network_call():
...

Async CSV Stream

Efficiently generate CSV data as a byte stream from an async generator of tuples.

frompython3_commons.generatorsimporttuple_csv_streamasyncdefgenerate_rows():
foriinrange(1000):
yield (i, f"Name {i}", 10.5*i)
asyncforchunkintuple_csv_stream(generate_rows(), header=("ID", "Name", "Value")):
# Send chunk to HTTP response or write to filepass