"""Canonical node-type and edge-type constants for the relationship graph. Adding a new edge type: 1. Add a ``REL_*`` constant here. 2. Add an entry to ``REL_DESCRIPTIONS`` (zh-CN, used by the frontend legend). 3. Wire a sync handler in ``sync_service.py`` that emits the edge. """ from __future__ import annotations # ── Node types ───────────────────────────────────────────────────── NODE_TYPE_PERSON = "person" NODE_TYPE_DEPARTMENT = "department" NODE_TYPE_HARDWARE_ASSET = "asset" NODE_TYPE_SOFTWARE = "software" NODE_TYPE_SOFTWARE_SEAT = "seat" NODE_TYPE_API_KEY = "api_key" NODE_TYPE_PERMISSION = "permission" NODE_TYPE_EMAIL_GROUP = "email_group" NODE_TYPE_CLOUD_SERVER = "cloud_server" NODE_TYPE_CLOUD_SCOPE = "cloud_scope" NODE_TYPE_CLOUD_GRANT = "cloud_grant" NODE_TYPE_DOMAIN = "domain" NODE_TYPE_DNS_RECORD = "dns_record" NODE_TYPE_INTEGRATION = "integration" ALL_NODE_TYPES = frozenset({ NODE_TYPE_PERSON, NODE_TYPE_DEPARTMENT, NODE_TYPE_HARDWARE_ASSET, NODE_TYPE_SOFTWARE, NODE_TYPE_SOFTWARE_SEAT, NODE_TYPE_API_KEY, NODE_TYPE_PERMISSION, NODE_TYPE_EMAIL_GROUP, NODE_TYPE_CLOUD_SERVER, NODE_TYPE_CLOUD_SCOPE, NODE_TYPE_CLOUD_GRANT, NODE_TYPE_DOMAIN, NODE_TYPE_DNS_RECORD, NODE_TYPE_INTEGRATION, }) # ── Edge types ───────────────────────────────────────────────────── # Person ↔ Person REL_MANAGES = "manages" # manager -> subordinate REL_REPORTS_TO = "reports_to" # subordinate -> manager (mirror) # Person ↔ Department REL_MEMBER_OF = "member_of" # person -> department REL_LEADS = "leads" # person -> department (manager-of-leaf) # Person ↔ Asset REL_ASSIGNED_TO = "assigned_to" # asset -> person (asset assigned) REL_HOLDS_SEAT = "holds_seat" # person -> seat REL_HAS_API_KEY = "has_api_key" # person -> api_key REL_OWNS = "owns" # person -> resource (generic ownership) REL_RESPONSIBLE_FOR = "responsible_for" # person -> software_catalog (business owner) REL_IT_CONTACT_FOR = "it_contact_for" # person -> software_catalog (IT contact) REL_GRANTS_ACCESS_TO = "grants_access_to" # cloud_grant -> cloud_scope # Asset ↔ Asset REL_HOSTS = "hosts" # cloud_server -> software (hosted on) REL_RESOLVES_DNS = "resolves_dns" # domain -> cloud_server / dns_record ALL_EDGE_TYPES = frozenset({ REL_MANAGES, REL_REPORTS_TO, REL_MEMBER_OF, REL_LEADS, REL_ASSIGNED_TO, REL_HOLDS_SEAT, REL_HAS_API_KEY, REL_OWNS, REL_RESPONSIBLE_FOR, REL_IT_CONTACT_FOR, REL_GRANTS_ACCESS_TO, REL_HOSTS, REL_RESOLVES_DNS, }) # ── Sync sources ─────────────────────────────────────────────────── SOURCE_IDENTITY = "sync.identity" SOURCE_ASSET = "sync.asset" SOURCE_SEAT = "sync.seat" SOURCE_API_KEY = "sync.api_key" SOURCE_PERMISSION = "sync.permission" SOURCE_EMAIL_GROUP = "sync.email_group" SOURCE_SOFTWARE = "sync.software" SOURCE_CMDB = "sync.cmdb" SOURCE_MANUAL = "manual" # Edges (rel_type, source) tuples are how we identify which sync # is responsible for which class of edge. A sync handler operating # on a given source SOFT-DELETES (sets ``deleted_at``) edges of its # source that disappear from the source data and UPSERTS edges still # present. Other sources' edges on the same vertices are untouched. # Maps each source to the (rel_type, from_type, to_type) tuples it owns. SOURCE_OWNS: dict[str, list[tuple[str, str, str]]] = { SOURCE_IDENTITY: [ (REL_MANAGES, NODE_TYPE_PERSON, NODE_TYPE_PERSON), (REL_REPORTS_TO, NODE_TYPE_PERSON, NODE_TYPE_PERSON), (REL_MEMBER_OF, NODE_TYPE_PERSON, NODE_TYPE_DEPARTMENT), ], SOURCE_ASSET: [ (REL_ASSIGNED_TO, NODE_TYPE_HARDWARE_ASSET, NODE_TYPE_PERSON), ], SOURCE_SEAT: [ (REL_HOLDS_SEAT, NODE_TYPE_PERSON, NODE_TYPE_SOFTWARE_SEAT), (REL_OWNS, NODE_TYPE_SOFTWARE_SEAT, NODE_TYPE_SOFTWARE), ], SOURCE_API_KEY: [ (REL_HAS_API_KEY, NODE_TYPE_PERSON, NODE_TYPE_API_KEY), ], SOURCE_PERMISSION: [ (REL_OWNS, NODE_TYPE_PERSON, NODE_TYPE_PERMISSION), ], SOURCE_EMAIL_GROUP: [ (REL_OWNS, NODE_TYPE_PERSON, NODE_TYPE_EMAIL_GROUP), ], SOURCE_SOFTWARE: [ (REL_RESPONSIBLE_FOR, NODE_TYPE_PERSON, NODE_TYPE_SOFTWARE), (REL_IT_CONTACT_FOR, NODE_TYPE_PERSON, NODE_TYPE_SOFTWARE), ], SOURCE_CMDB: [ (REL_OWNS, NODE_TYPE_PERSON, NODE_TYPE_CLOUD_SERVER), (REL_OWNS, NODE_TYPE_PERSON, NODE_TYPE_CLOUD_SCOPE), (REL_OWNS, NODE_TYPE_PERSON, NODE_TYPE_DOMAIN), (REL_GRANTS_ACCESS_TO, NODE_TYPE_PERSON, NODE_TYPE_CLOUD_SCOPE), ], }