Agent Zero
119 lines
3.7 KiB
Python
119 lines
3.7 KiB
Python
"""Public contract for the contacts plugin.
|
|
|
|
Exposes the symbols that other core modules and plugins need without
|
|
importing from internal modules directly (Block C7: dashboard counts).
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
from typing import Any
|
|
|
|
from sqlalchemy import func, select
|
|
from sqlalchemy.ext.asyncio import AsyncSession
|
|
|
|
from app.core.visibility import apply_visibility_filter
|
|
from app.models.contact import Contact
|
|
from app.plugins.builtins.contracts import get_contract_registry
|
|
|
|
|
|
class ContactsContract:
|
|
"""Public API surface for the contacts plugin."""
|
|
|
|
contract_name = "contacts"
|
|
|
|
@staticmethod
|
|
async def get_counts(
|
|
db: AsyncSession,
|
|
tenant_id: Any,
|
|
user_id: Any,
|
|
is_system_admin: bool = False,
|
|
) -> dict[str, int]:
|
|
"""Return visibility-filtered contact/company/person counts."""
|
|
queries = []
|
|
for type_filter in (None, "company", "person"):
|
|
query = select(func.count(Contact.id)).where(
|
|
Contact.tenant_id == tenant_id,
|
|
Contact.deleted_at.is_(None),
|
|
)
|
|
if type_filter is not None:
|
|
query = query.where(Contact.type == type_filter)
|
|
query = await apply_visibility_filter(
|
|
db, query, "contact", Contact, user_id, tenant_id, is_system_admin
|
|
)
|
|
queries.append(query)
|
|
|
|
results = [((await db.execute(q)).scalar() or 0) for q in queries]
|
|
return {
|
|
"contacts": results[0],
|
|
"companies": results[1],
|
|
"persons": results[2],
|
|
"total": results[0],
|
|
}
|
|
|
|
@staticmethod
|
|
async def dsar_collect(
|
|
db: AsyncSession, tenant_id: Any, user_id: Any
|
|
) -> dict[str, Any]:
|
|
"""GDPR Art. 15: collect contact data owned by the user.
|
|
|
|
Owned by the contacts plugin — core/jobs.py calls this generically
|
|
via the contract, it must not know contact internals.
|
|
"""
|
|
contacts = (
|
|
await db.execute(
|
|
select(Contact).where(
|
|
Contact.tenant_id == tenant_id,
|
|
Contact.owner_id == user_id,
|
|
Contact.deleted_at.is_(None),
|
|
)
|
|
)
|
|
).scalars().all()
|
|
return {
|
|
"contacts": [
|
|
{
|
|
"id": str(c.id),
|
|
"type": c.type,
|
|
"displayname": c.displayname,
|
|
"email_1": c.email_1,
|
|
"email_2": c.email_2,
|
|
}
|
|
for c in contacts
|
|
]
|
|
}
|
|
|
|
@staticmethod
|
|
async def dsar_erase(
|
|
db: AsyncSession, tenant_id: Any, user_id: Any
|
|
) -> dict[str, int]:
|
|
"""GDPR Art. 17: soft-delete contacts owned by the user.
|
|
|
|
Soft-delete via deleted_at (audit history must remain intact — it is
|
|
a business record, not personal data of the subject; retention
|
|
policy governs its cleanup).
|
|
"""
|
|
from datetime import UTC, datetime
|
|
|
|
contacts = (
|
|
await db.execute(
|
|
select(Contact).where(
|
|
Contact.tenant_id == tenant_id,
|
|
Contact.owner_id == user_id,
|
|
Contact.deleted_at.is_(None),
|
|
)
|
|
)
|
|
).scalars().all()
|
|
for c in contacts:
|
|
c.deleted_at = datetime.now(UTC)
|
|
return {"contacts_soft_deleted": len(contacts)}
|
|
|
|
@classmethod
|
|
def get_function(cls, name: str):
|
|
"""Return a callable exposed by this contract, or None if absent."""
|
|
return getattr(cls, name, None)
|
|
|
|
|
|
# ─── self-registration ───
|
|
|
|
_contract = ContactsContract()
|
|
get_contract_registry().register("contacts", _contract)
|