morpheus-data-layer

Import prefix morpheus_data_layer. Layer MorpheusDataLayer. Used by all five Lambda functions; not used by the AgentCore runtime, which has no database access.

Source: libs/morpheus-data-layer/morpheus_data_layer/.

The schema is owned by the Payability web application’s Prisma schema; this library is a read/write ORM over a subset of it. See Prisma schema reference.

session.py

A module-level engine singleton, so one connection pool serves the whole process — whether that is a warm Lambda container, the AgentCore runtime, or a script.

def init(database_url: str, *, pool_size: int = 5, max_overflow: int = 2) -> None
@contextmanager
def get_session() -> Generator[Session, None, None]
def check_connection() -> bool

init() is idempotent: a second call is a no-op and the existing engine and pool are kept intact.

The layer installs psycopg2-binary only. init() therefore always names the driver: postgresql://, postgres://, and postgresql+psycopg:// are rewritten to postgresql+psycopg2:// (the last with a warning). Without this, SQLAlchemy 2.1+ resolves a bare postgresql:// to psycopg 3 and the cold start fails with No module named 'psycopg'. SQLAlchemy is also pinned to >=2.0,<2.1 in requirements.txt, the file SAM installs the layer from, so the layer matches the version in uv.lock that the tests run against.

Pooling

pool_size Behaviour

0

NullPool — a fresh connection per get_session() block, released on exit

> 0

QueuePool with pool_pre_ping=True, pool_recycle=300, and max_overflow

Always pool_size=0 in a Lambda. Every function calls init(_secrets.database_url, pool_size=0) at cold start.

Each Lambda container handles one invocation at a time and gains nothing from a persistent pool, while NullPool lets Supabase’s PgBouncer multiplex efficiently across hundreds of concurrent containers. A persistent pool per container would exhaust the connection limit instead.

Equally: use the Supabase Transaction mode pooler URL (port 6543), not Session mode (port 5432). Session mode reserves one server connection per client for its entire lifetime.

Every new connection runs SET search_path TO public through a connect event listener.

Transaction boundary

with get_session() as session:
    seller = _seller_repo.get_by_seller_id(session, seller_id)
    _seller_repo.update(session, seller, {"lifeCycleStage": SellerLifeCycleStage.ENRICHED})
# committed here

Commits on clean exit; rolls back and re-raises on any exception; always closes. The session factory uses expire_on_commit=False, so scalar attributes stay readable on a detached object after the block ends.

Two consequences of expire_on_commit=False worth internalising:

  • Scalars survive the block. Relationships and unloaded columns do not — accessing them after commit raises. This is why _build_crm_sync_event() in enrichment-scorer must run inside the session, and why _upsert_seller() explicitly touches enrichedAt, payoutPainScore, and note before the block ends.

  • Handlers that call an external API between reads and writes deliberately use two short sessions rather than one long one, so a connection is not held open across network latency. `email-agent’s categorisation handler is the clearest example.

check_connection() runs SELECT 1 and returns a boolean — for health checks.

models/

One SQLAlchemy 2.0 declarative model per file, all on a shared Base, exported from models/init.py.

Model Table

Seller

sellers

Contact

contacts

Activity

Activity

Call

Call

CampaignTarget

CampaignTarget

Configs

configs

Email

emails

EmailCampaignTarget

email_campaign_targets

EmailVariation

email_variations

FollowUpTask

follow_up_tasks

SalesRep

sales_reps

TargetEmails

target_emails

TargetPhoneNumber

target_phone_numbers

Conventions:

  • Attribute names are camelCase and match the database column exactly. No name= override in mapped_column — the Python and SQL names are identical, which is unusual for SQLAlchemy but keeps conflict_columns lists and raw SQL honest.

  • ForeignKey("subscriptions.id") is not declared. That table is not managed here; the constraint is enforced at the database level only.

  • One model per file, always exported from models/init.py.

Details: ORM models.

repositories/

Stateless CRUD objects, instantiated once at module level by consumers:

_seller_repo = SellerRepository()      # module level — stateless, safe to share

They never open sessions and never commit. Every method takes a Session as its first argument.

BaseRepository[ModelT]

Method Behaviour

get_by_id(session, record_id)

session.get() — None if missing

get_all(session, *, limit=100, offset=0)

Paged select

filter_by(session, **kwargs)

Equality filters, all matches

first_by(session, **kwargs)

Equality filters, first match

insert(session, data)

Add, flush, refresh, return the instance

bulk_insert(session, rows)

Core insert() for many rows

upsert(session, data, conflict_columns, update_columns=None)

PostgreSQL ON CONFLICT DO UPDATE with RETURNING; update_columns defaults to every non-conflict key in data

update(session, record, data)

setattr each field, flush, refresh

update_by_id(session, record_id, data)

Fetch then update; None if missing

conflict_columns and update_columns take camelCase database column names, matching the ORM attributes.

Concrete repositories

Each adds query methods for the access patterns the handlers actually need. Notable ones:

Repository Domain-specific methods

SellerRepository

get_by_seller_id, get_pending_enrichment, get_pending_scoring, get_by_platform, get_suspended, upsert_by_seller_id

ContactRepository

get_by_email, get_by_phone, get_by_hubspot_contact_id, get_by_seller, list_missing_hubspot_contact_id, list_missing_hubspot_id_by_last_sync, find_existing, upsert_for_seller

FollowUpTaskRepository

list_pending_due_soon, list_pending_with_hubspot_task, has_active_task_for_contact, list_stale_for_escalation

SalesRepRepository

get_active_by_email, get_active_by_hubspot_owner_id, get_by_hubspot_owner_id, list_active_for_subscription

CallRepository

get_by_provider_call_id, get_with_target_for_subscription, get_with_target_by_ids

ConfigsRepository

get_by_key, get_value, get_by_subscription, set_value

TargetPhoneNumberRepository

get_by_phone_number, list_pending_twilio_sync

TargetEmailsRepository

get_by_subscription_and_email, set_newsletter_opt_out

EmailCampaignTargetRepository

get_by_campaign_and_email, update_categorisation

ActivityRepository

get_by_external_id, list_by_contact

EmailRepository and EmailVariationRepository add nothing beyond the base.

Full signatures: Repositories.

Round-robin selection

repositories/sales_rep_round_robin.py provides select_next_sales_rep_round_robin(), reserve_sales_rep_round_robin_batch() and dialer_rr_config_key(). These are functions, not repository methods, because they span sales_reps and configs and take a PostgreSQL advisory transaction lock. See Ownership and round-robin.

reserve_sales_rep_round_robin_batch(session, subscription_id, , require_nooks, count) reserves count reps in one call, advancing the cursor once. Prefer it when assigning many records at once: select_next_sales_rep_round_robin() (which now delegates to it with count=1) costs a rep query, an advisory lock and a cursor write *per call, so calling it in a loop over thousands of rows will exhaust a Lambda timeout.

enums.py

StrEnum mirrors of the Prisma/PostgreSQL enums, so a comparison against a string literal works but the canonical member is available: SellerPlatform, SellerDatasource, SellerLifeCycleStage, ContactHubspotSyncStatus, ContactSource, ContactHeatStatus, ContactLifecycleStage, ContactLeadStatus, EmailTargetStatus, EmailTargetCategory, EmailDirection, EmailInboxCategory, EmailVariationStatus, LineTypes, RiskLevel, CallStatus, TargetState, FollowUpTaskType, FollowUpTaskStatus, ActivityType.

Note LineTypes uses camelCase values (fixedVoip, nonFixedVoip, tollFree) because they mirror Twilio’s vocabulary exactly. Full listing: Enumerations.

Adding a model or repository

  1. Create models/<name>.py with camelCase attributes matching the columns exactly. No subscriptions foreign key.

  2. Export it from models/init.py.

  3. Create repositories/<name>.py extending BaseRepository[YourModel] with model_class = YourModel.

  4. Add only the query methods the callers need. No sessions, no commits.

  5. Export it from repositories/init.py.

  6. Add a test under tests/.

Tests

cd libs/morpheus-data-layer && uv run pytest tests -v

tests/integration/ (test_connection.py, test_seller_repository.py, test_configs_repository.py) needs a reachable database and reads its connection string from libs/morpheus-data-layer/.env. A Makefile in the library directory wraps the common commands.