Repositories

Repositories are the only sanctioned way to reach the database. They live in libs/morpheus-data-layer/morpheus_data_layer/repositories/, one file per model, and every public name is re-exported from repositories/init.py.

Two rules govern all of them:

Repositories are stateless

Instantiate once at module level — _repo = SellerRepository() — and reuse across invocations. There is no per-request construction cost and no instance state to leak between Lambda invocations.

Repositories never own a transaction

Every method takes an explicit Session as its first positional argument. No method calls commit(), rollback(), or opens a session. The caller does that with with get_session() as session:, which commits on a clean exit and rolls back on an exception. A handler that calls three repositories inside one with block gets one atomic transaction across all three.

from morpheus_data_layer.repositories import SellerRepository, ContactRepository
from morpheus_data_layer.session import get_session

_seller_repo = SellerRepository()
_contact_repo = ContactRepository()

with get_session() as session:
    seller = _seller_repo.upsert_by_seller_id(session, data)
    _contact_repo.upsert_for_seller(session, contact_data)
    # commit happens here, once, for both writes

BaseRepository

repositories/base.py — a generic BaseRepository[ModelT]. A subclass binds it to a model with a single class attribute:

class SellerRepository(BaseRepository[Seller]):
    model_class = Seller
Method Behaviour

get_by_id(session, record_id)

Primary-key lookup. Returns None when absent.

get_many_by_ids(session, record_ids)

WHERE pk IN (…​). Raises ValueError on a composite primary key.

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

Paginated full scan.

filter_by(session, **kwargs)

Equality filters on mapped attributes. Returns a list.

first_by(session, **kwargs)

Same filters, first match or None.

insert(session, data)

add + flush + refresh. Returns the populated instance; caller commits.

bulk_insert(session, rows)

Multi-row insert with no ORM instances returned.

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

PostgreSQL INSERT … ON CONFLICT DO UPDATE, returning the row via RETURNING.

update(session, record, data)

setattr each key, then flush + refresh.

update_by_id(session, record_id, data)

Fetch then update. Returns None if the row does not exist.

BaseRepository has no delete and no count. Morpheus does not delete rows — records are soft-signalled through status columns (isActive, isDeleted, lifecycle stages) so that history survives. If you need a count, write a scalar query in the concrete repository.

The upsert contract

upsert() is how every idempotent write is done, because SQS delivers at-least-once and schedules overlap.

repo.upsert(
    session,
    data,
    conflict_columns=["sellerId"],                     (1)
    update_columns=["businessName", "enrichedAt"],     (2)
)
1 Must match a real unique constraint or unique index, written as the camelCase database column name.
2 Optional. When omitted, every key in data except the conflict columns is updated.

Because ORM attribute names are identical to column names (see ORM models), these lists are literal SQL identifiers — grep finds every place a column participates in a conflict resolution.

Repository reference

SellerRepository

repositories/seller.py → Seller

Method Behaviour

get_by_seller_id(session, seller_id)

Lookup by the external marketplace id, not the primary key.

get_by_platform(session, platform, *, limit=100, offset=0)

Filter on SellerPlatform, newest createdAt first.

get_by_subscription(session, subscription_id, *, limit=100, offset=0)

Tenant-scoped listing, newest first.

get_pending_enrichment(session, *, limit=10)

lifeCycleStage == NEW and enrichedAt IS NULL, oldest first.

get_pending_scoring(session, *, limit=10)

enrichedAt IS NOT NULL and payoutPainScore IS NULL, oldest enrichedAt first.

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

isSuspended is true.

upsert_by_seller_id(session, data)

Upsert on sellerId, updating every key in data except sellerId, id, createdAt.

The two get_pending_* methods are the queue for the backfill schedules — they define what "pending" means for enrichment and scoring, and the ordering is what makes the backfill fair rather than starving the oldest rows.

ContactRepository

repositories/contact.py → Contact

Method Behaviour

get_by_email(session, email)

Exact match on the nullable-unique email column.

get_by_phone(session, phone)

Exact match on the nullable-unique phone column.

get_by_hubspot_contact_id(session, hubspot_id)

Match on the CRM identifier.

get_by_seller(session, seller_id, *, limit=100, offset=0)

All contacts of one seller, newest first.

list_missing_hubspot_contact_id(session, *, subscription_id, limit)

Contacts with no hubspotContactId that have an email, phone, or mobile phone. Oldest first.

list_missing_hubspot_id_by_last_sync(session, *, subscription_id, limit=25)

Same population, ordered hubspotSyncPriority DESC, then hubspotLastSyncedAt ASC NULLS FIRST.

find_existing(session, *, email, phone, hubspot_contact_id)

Identity resolution. Tries hubspotContactId, then email, then phone, in that order.

upsert_for_seller(session, data)

Find-or-create with merge semantics — see below.

upsert_for_seller() is not a PostgreSQL upsert. It resolves an existing contact through find_existing(), then merges — and it will never overwrite a non-null column with NULL, and it strips an incoming email or phone that would collide with a different contact row.

That behaviour exists because Contact.email and Contact.phone are nullable-unique globally. A naive ON CONFLICT would either raise on the collision or silently blank out a verified phone number with an empty scrape result. The merge rule is the whole reason enrichment can run repeatedly against the same contact without degrading it.

The two list_missing_* methods differ only in ordering, and both exist on purpose: Hubspot.ContactResolve uses the priority-ordered one so that high-value contacts reach the CRM first, while the age-ordered one guarantees nothing is stranded forever. See CRM sync.

ActivityRepository

repositories/activity.py → Activity

Method Behaviour

get_by_external_id(session, *, subscription_id, external_id)

Lookup on the unique (subscriptionId, externalId) pair.

list_by_contact(session, contact_id, *, activity_type=None, limit=100, offset=0)

Timeline for one contact, newest first, optionally filtered by ActivityType.

Activity upserts are performed by the caller rather than the repository. The BigQuery engagement sync calls the base upsert() directly with:

conflict_columns=["subscriptionId", "externalId"]
update_columns=["type", "hubspotEngagementId", "subject", "body", "status", "direction",
                "contactId", "contactEmail", "hubspotContactId", "metadata", "updatedAt"]

externalId is the HubSpot engagement id, which is what makes re-running the sync over an overlapping watermark window free of duplicates. See Scheduled events.

CallRepository

repositories/call.py → Call

Method Behaviour

get_by_provider_call_id(session, provider_call_id)

Lookup by the dialer’s own call identifier.

get_with_target_for_subscription(session, call_id, subscription_id)

Tenant-scoped fetch with joinedload(Call.target).

get_with_target_by_ids(session, call_id, target_id, subscription_id)

Same, but also asserts the target id — defence in depth against a mismatched pair in a webhook.

The Five9 sync upserts through the base method on conflict_columns=["providerCallId"], updating targetPhoneNumber, direction, status, disposition, durationMs, durationSec, startTimestamp, hubspotEngagementId, hubspotSyncStatus, contactName, notes, targetPhoneNumbersId, and salesRepId. See Call logging and dispositions.

SalesRepRepository

repositories/sales_rep.py → SalesRep

Method Behaviour

get_active_by_email(session, email, subscription_id)

Active rep matching an email address.

get_active_by_hubspot_owner_id(session, hubspot_owner_id, subscription_id)

Active rep matching a HubSpot owner id.

get_by_hubspot_owner_id(session, hubspot_owner_id, subscription_id)

Same lookup without the isActive filter.

list_active_for_subscription(session, subscription_id, *, require_hubspot=True, require_nooks=False)

Active reps, ordered id ASC, optionally requiring a HubSpot owner id or a Nooks id.

get_by_hubspot_owner_id() has a docstring that says "active", but the query carries no isActive predicate. It resolves a rep for an inbound record — a call or engagement already owned by someone who may since have been deactivated — where dropping the row would lose history. get_active_by_hubspot_owner_id() is the one to use when assigning new work.

The id ASC ordering in list_active_for_subscription() is load-bearing: it is the stable ordering that makes the round-robin cursor below meaningful.

Round-robin helpers

repositories/sales_rep_round_robin.py. Functions, not a class, because the state they manage lives in the configs table rather than in a model.

Name Behaviour

RR_CONFIG_KEY_PREFIX

The constant "dialer_follow_up_rep_rr".

dialer_rr_config_key(subscription_id)

Builds dialer_follow_up_rep_rr:<subscriptionId>.

reserve_sales_rep_round_robin_batch(session, subscription_id, *, require_nooks, count, sales_rep_repo=None, configs_repo=None)

Reserves count reps and advances the cursor by count. Returns [] when no rep qualifies.

select_next_sales_rep_round_robin(session, subscription_id, *, require_nooks, sales_rep_repo=None, configs_repo=None)

count=1 convenience wrapper.

The batch function takes a PostgreSQL transaction-level advisory lock before reading the cursor. Without it, two concurrent re-engagement handlers would read the same cursor value and assign the same rep to both batches, defeating the point of round-robin. The lock is released when the transaction ends, so it must be held inside the caller’s with get_session() block.

Reserving a batch and advancing once — rather than calling the single-rep helper in a loop — keeps the lock window short and the distribution even. Full rules in Ownership and round-robin.

ConfigsRepository

repositories/configs.py → Configs

Method Behaviour

get_by_key(session, key)

Row lookup on the globally unique key.

get_value(session, key)

The value string, or None.

get_by_subscription(session, subscription_id, *, limit=100, offset=0)

All rows for one tenant.

set_value(session, key, value, subscription_id)

Upsert with conflict_columns=["key"], update_columns=["value"].

Every key in use is catalogued in Runtime state in configs.

FollowUpTaskRepository

repositories/follow_up_task.py → FollowUpTask

Method Behaviour

list_pending_due_soon(session, subscription_id, window_hours=24)

PENDING tasks due within the window. Eager-loads salesRep.

list_pending_with_hubspot_task(session, subscription_id, limit=50)

PENDING or FAILED tasks that carry a hubspotTaskId, FAILED first.

has_active_task_for_contact(session, *, contact_id, subscription_id, task_type=FollowUpTaskType.CALL)

True when a PENDING or IN_PROGRESS task exists that is not past due.

list_stale_for_escalation(session, subscription_id)

PENDING, due at least a day ago, escalationStage < 3. Eager-loads salesRep and contact.

has_active_task_for_contact() is the deduplication guard every re-engagement handler calls before creating work — it is what stops four segments from putting the same contact on a rep’s list four times. The escalationStage < 3 ceiling caps escalation at three notifications. See Follow-up tasks and escalation.

The eager loads matter: expire_on_commit=False keeps scalars readable after commit but raises on an unloaded relationship, so a handler that formats a Slack message from task.salesRep.email after the session closes needs the relationship already loaded.

Email repositories

Repository Model Methods beyond BaseRepository

EmailRepository

Email

None — a plain model wrapper.

EmailVariationRepository

EmailVariation

None — a plain model wrapper.

EmailCampaignTargetRepository

EmailCampaignTarget

get_by_campaign_and_email(session, *, campaign_id, email) — case-insensitive email match.
update_categorisation(session, record, *, category, categorised_at, category_model) — writes category, categorisedAt, categoryModel.

TargetEmailsRepository

TargetEmails

get_by_subscription_and_email(session, *, subscription_id, email) — case-insensitive.
set_newsletter_opt_out(session, *, subscription_id, email, opted_out, occurred_at) — guarded write.

update_categorisation() records category_model alongside the verdict, so a misclassification can be traced back to the model that produced it — see email-agent.

set_newsletter_opt_out() compares occurred_at against the stored newsletterOptOutAt and ignores the write when the incoming event is older. HubSpot webhooks arrive out of order, and without the guard a stale "subscribed" event could silently undo a later opt-out — a compliance problem, not just a data problem.

TargetPhoneNumberRepository

repositories/target_phone_number.py → TargetPhoneNumber

Method Behaviour

get_by_phone_number(session, phone_number)

Substring match (phoneNumber.contains(…​)), tolerating format differences between sources.

list_pending_twilio_sync(session, *, subscription_id, limit)

syncTwilio false, active, not deleted. Ordered syncPriority DESC, then createdAt ASC.

list_pending_twilio_sync() is the work queue for phone line intelligence; the batch size comes from PHONE_AGENT_BATCH_SIZE, and each Twilio Lookup call costs money, so the priority ordering decides what the budget is spent on.

Exports

repositories/init.py exports exactly:

ActivityRepository, BaseRepository, CallRepository, ConfigsRepository,
ContactRepository, EmailRepository, EmailCampaignTargetRepository,
EmailVariationRepository, FollowUpTaskRepository, SalesRepRepository,
SellerRepository, TargetEmailsRepository, TargetPhoneNumberRepository,
dialer_rr_config_key, reserve_sales_rep_round_robin_batch,
select_next_sales_rep_round_robin

There is no CampaignTargetRepository. CampaignTarget is mapped as a model and written through the base upsert() by morpheus-hubspot directly.

Adding a repository

  1. Create repositories/<name>.py with class XRepository(BaseRepository[X]) and model_class = X.

  2. Add only the queries a handler actually needs. Ad-hoc filtering belongs in filter_by().

  3. Take session: Session first, positionally; make everything else keyword-only.

  4. Never commit, never open a session, never catch and swallow a database error.

  5. Use camelCase column names in conflict_columns and update_columns.

  6. Export it from repositories/init.py.

  7. Update this page and morpheus-data-layer.