"""The tables, and the code that turns rows into contacts and activities and back."""
from datetime import date, datetime, timezone
from typing import Optional
from sqlalchemy import (BigInteger, Boolean, Column, Date, Integer, MetaData, String, Table, TIMESTAMP, and_,
delete, insert, or_, select)
from sqlalchemy.exc import IntegrityError
from erp_platform.repository import Aggregate, Part
from brick_activity.domain.activity import Activity, ActivityRuleBroken, Contact
md = MetaData()
contacts = Table("activity_contacts", md,
Column("contact_id", String, primary_key=True), Column("bp_id", BigInteger),
Column("full_name", String), Column("email", String), Column("owner", String),
Column("title", String), Column("phone", String), Column("subscribed", Boolean),
Column("status", String), Column("created_at", TIMESTAMP), Column("updated_at", TIMESTAMP),
Column("version", Integer)) # 0.2.0: the optimistic version
emails = Table("activity_contact_emails", md,
Column("email", String, primary_key=True), Column("contact_id", String),
Column("created_at", TIMESTAMP))
log = Table("activity_log", md,
Column("activity_id", String, primary_key=True), Column("kind", String), Column("subject", String),
Column("owner", String), Column("bp_id", BigInteger), Column("contact_id", String),
Column("opportunity_id", String), Column("note", String), Column("direction", String),
Column("occurred_at", TIMESTAMP), Column("due_on", Date), Column("status", String),
Column("done_at", TIMESTAMP), Column("created_at", TIMESTAMP), Column("updated_at", TIMESTAMP),
Column("version", Integer)) # 0.2.0: the optimistic version
CONTACT = Aggregate(Part(contacts, ("contact_id",)), version="version")
ACTIVITY = Aggregate(Part(log, ("activity_id",)), version="version")
class EmailTaken(ActivityRuleBroken):
pass
def _iso(v) -> Optional[str]:
return v.isoformat() if hasattr(v, "isoformat") else (v or None)
def _ts(v) -> Optional[datetime]:
if not v:
return None
d = v if isinstance(v, datetime) else datetime.fromisoformat(str(v).replace("Z", "+00:00"))
return d if d.tzinfo else d.replace(tzinfo=timezone.utc)
def _d(v) -> Optional[date]:
return None if not v else (v if isinstance(v, date) and not isinstance(v, datetime)
else date.fromisoformat(str(v)[:10]))
def _is_duplicate(exc: Exception) -> bool:
"""A primary-key collision, however the driver says it: IntegrityError (psycopg, SQLite) or SQLSTATE
23505 in the message (the CLI the live check runs through)."""
text = str(exc).lower()
return isinstance(exc, IntegrityError) or "23505" in text or "duplicate key" in text
def lost_email_claim(exc: BaseException) -> bool:
"""This unit of work lost the race for an email: the claim was refused at the INSERT (EmailTaken) or —
DataK3 checks a staged key against another transaction's at COMMIT — as a duplicate on
activity_contact_emails when the transaction committed, outside the repository."""
seen = set()
while exc is not None and id(exc) not in seen:
seen.add(id(exc))
if isinstance(exc, EmailTaken):
return True
text = str(exc).lower()
if _is_duplicate(exc) and "activity_contact_emails" in text:
return True
exc = getattr(exc, "orig", None) or exc.__cause__ or exc.__context__
return False
class ActivityRepository:
def __init__(self, conn):
self.conn = conn
# ── contacts ─────────────────────────────────────────────────────────────────────────
def _contact(self, r) -> Contact:
return Contact(r["contact_id"], int(r["bp_id"]), r["full_name"], r["email"], r["owner"],
r["title"] or "", r["phone"] or "", bool(r["subscribed"]), r["status"],
int(r.get("version") or 0))
def contact(self, contact_id: str) -> Optional[Contact]:
rows = CONTACT.load(self.conn, {"contact_id": contact_id})
return None if rows is None else self._contact(rows["activity_contacts"][0])
def contact_by_email(self, email: str) -> Optional[Contact]:
r = self.conn.execute(select(emails.c.contact_id).where(emails.c.email == email)).first()
return None if r is None else self.contact(r[0])
def contacts_at(self, bp_id: int) -> list:
rows = self.conn.execute(select(contacts).where(contacts.c.bp_id == int(bp_id)))
return sorted((self._contact(dict(r._mapping)) for r in rows), key=lambda c: c.full_name.lower())
def claim_email(self, email: str, contact_id: str) -> None:
"""A plain INSERT on purpose: the one write here that must FAIL on a duplicate, not update it."""
try:
self.conn.execute(insert(emails).values(email=email, contact_id=contact_id,
created_at=datetime.now(timezone.utc)))
except Exception as e: # noqa: BLE001 — re-raised below
if _is_duplicate(e):
raise EmailTaken(f"{email} already belongs to another contact") from None
raise
def release_email(self, email: str, contact_id: str) -> None:
self.conn.execute(delete(emails).where(and_(emails.c.email == email, emails.c.contact_id == contact_id)))
def save_contact(self, c: Contact) -> None:
"""Claims version n+1 against the version the contact was loaded at (0 for a new one)."""
c.version = CONTACT.save(self.conn, {"activity_contacts": [{
"contact_id": c.contact_id, "bp_id": c.bp_id, "full_name": c.full_name, "email": c.email,
"owner": c.owner, "title": c.title, "phone": c.phone, "subscribed": c.subscribed,
"status": c.status}]}, expected_version=c.version)
# ── activities ───────────────────────────────────────────────────────────────────────
def _activity(self, r) -> Activity:
return Activity(r["activity_id"], r["kind"], r["subject"], r["owner"], int(r["bp_id"]), r["contact_id"],
r["opportunity_id"], r["note"] or "", r["direction"], _iso(_ts(r["occurred_at"])),
_iso(_d(r["due_on"])), r["status"], _iso(_ts(r["done_at"])), int(r.get("version") or 0))
def activity(self, activity_id: str) -> Optional[Activity]:
rows = ACTIVITY.load(self.conn, {"activity_id": activity_id})
return None if rows is None else self._activity(rows["activity_log"][0])
def save_activity(self, a: Activity) -> None:
"""Claims version n+1 against the version the activity was loaded at (0 for a new one)."""
a.version = ACTIVITY.save(self.conn, {"activity_log": [{
"activity_id": a.activity_id, "kind": a.kind, "subject": a.subject, "owner": a.owner,
"bp_id": a.bp_id, "contact_id": a.contact_id, "opportunity_id": a.opportunity_id, "note": a.note,
"direction": a.direction, "occurred_at": _ts(a.occurred_at), "due_on": _d(a.due_on),
"status": a.status, "done_at": _ts(a.done_at)}]}, expected_version=a.version)
def timeline(self, *, contact_id: Optional[str] = None, bp_id: Optional[int] = None,
opportunity_id: Optional[str] = None, limit: int = 100) -> list:
"""Newest first (CRM-02). For a company, its contacts' activities are its own — they work there."""
conds = []
if contact_id:
conds.append(log.c.contact_id == contact_id)
if bp_id is not None:
conds.append(log.c.bp_id == int(bp_id))
if opportunity_id:
conds.append(log.c.opportunity_id == opportunity_id)
rows = self.conn.execute(select(log).where(or_(*conds) if len(conds) > 1 else conds[0]))
acts = [self._activity(dict(r._mapping)) for r in rows]
return sorted(acts, key=lambda a: a.at, reverse=True)[:limit]
def tasks_of(self, owner: str) -> list:
rows = self.conn.execute(select(log).where(and_(log.c.owner == owner, log.c.kind == "TASK",
log.c.status == "open")))
return [self._activity(dict(r._mapping)) for r in rows]