Agent Zero
63 lines
1.9 KiB
Python
63 lines
1.9 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],
|
|
}
|
|
|
|
@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)
|