Shared Python utilities (common)

common is the shared Python 3.13 library used by backend services and jobs for Spanner access, structured logging, secrets, and shared models.

Package profile

What lives here

• common.spanner: Async Spanner wrapper, session pooling, transaction utilities, and index cache.
• common.spanner.models: Pydantic models for Spanner tables and mutation helpers.
• common.models: Cross-service request/response models (TaskInfo, RawEvent, PIMS/clinic/org, invoice webhook payloads).
• common.secrets: Secret Manager helpers with caching and CRC32c integrity checks.
• common.logging_utils: JSON logging formatter and request context helpers.
• common.cloudtasks: Async Cloud Tasks helper for queue inspection and de-duplication.
• common.security: Constant-time secret verification.
• common.util: Datetime normalization, gRPC timestamp helpers, and project ID resolution.

Spanner primitives

• Spanner: Async wrapper over SpannerAsyncClient with an AsyncBurstyPool session pool.
• TransactionManager: Async context manager that guarantees commit or rollback.
• SpannerModel: Base Pydantic model for Spanner rows with upsert_mutation and bulk_upsert helpers that handle commit timestamps and large batches.
• SpannerIndexCache: Caches secondary index counts for mutation sizing.
• SpannerModel.iter_models(): Useful for warming caches across all models.

Data model coverage

• Spanner row models for appointments, clinics, patients, organizations, invoices, pipelines, households, and Bubble mirror tables.
• Pydantic request/response models for webhook payloads and pipeline task metadata.

Logging and telemetry

• setup_logging configures JSON logs for Cloud Logging with JsonFormatter.
• request_context extracts HTTP metadata (method, path, task headers, trace id) for structured logs.
• Use extra={"extra_fields": {...}} to attach service-specific metadata without string formatting.

Secrets and security

• get_secret fetches from Secret Manager with a per-secret cache and CRC32c validation.
• get_service_sentry_dsn, get_service_api_key, get_raven_bubble_api_token wrap common lookups.
• Security.securely_verify_secret provides constant-time comparisons.

Cloud Tasks helpers

• CloudTasksHelper.list_tasks_for_queue lists tasks in a queue.
• CloudTasksHelper.is_task_scheduled checks for an existing TaskInfo-based task id.

Local development

• Install dependencies with uv sync --all-packages from repo root.
• No standalone runtime; validate changes by running the owning service tests.
• Keep imports absolute (import common...); the package ships with py.typed for type checking.

Configuration

• Requires ADC when accessing Spanner or Secret Manager.
• GCP_PROJECT_ID is used by resolve_project_id when running outside GCP.

Ownership

• Shared platform library; changes should be coordinated with service owners.

References

• Related: Spanner emulator setup