"""Abstract persistence API for 岗位 (positions) and their tag bindings.""" from __future__ import annotations from abc import ABC, abstractmethod from typing import Any class PositionStore(ABC): # -------------------------------------------------------------- positions @abstractmethod async def list_positions(self) -> list[dict[str, Any]]: """List all positions. Each record carries a derived ``member_count``.""" raise NotImplementedError @abstractmethod async def get_position(self, position_id: str) -> dict[str, Any] | None: raise NotImplementedError @abstractmethod async def create_position(self, data: dict[str, Any]) -> dict[str, Any]: """Create a position. ``data`` must carry ``id`` and ``name``. Raises ``ValueError`` if ``name`` is not unique. When ``is_default`` is truthy, any other default position is cleared first. """ raise NotImplementedError @abstractmethod async def update_position(self, position_id: str, data: dict[str, Any]) -> dict[str, Any] | None: """Update ``name`` / ``description`` / ``is_default``. Returns ``None`` if the position does not exist. Raises ``ValueError`` on a name collision. Setting ``is_default=True`` clears the flag on the other positions. """ raise NotImplementedError @abstractmethod async def delete_position(self, position_id: str) -> bool: """Delete a position, its tag bindings, and unassign its members.""" raise NotImplementedError @abstractmethod async def get_default_position_id(self) -> str | None: """Return the id of the ``is_default`` position, or ``None``.""" raise NotImplementedError # ------------------------------------------------------------ tag bindings @abstractmethod async def get_tag_bindings(self, position_id: str) -> dict[str, list[str]]: """Return ``{resource_type: [tag_id, ...]}`` for one position. Resource types with no bound tags are omitted (callers treat a missing key as "use the default set"). """ raise NotImplementedError @abstractmethod async def set_tag_bindings(self, position_id: str, bindings: dict[str, list[str]]) -> dict[str, list[str]]: """Replace the full set of tag bindings for a position. ``bindings`` maps each resource_type to its desired tag-id list. A resource_type omitted from ``bindings`` (or mapped to ``[]``) ends up with no rows — i.e. it falls back to the default set. """ raise NotImplementedError # --------------------------------------------------------------- members @abstractmethod async def get_user_position_id(self, user_id: str) -> str | None: """Return the position id explicitly assigned to a user, or ``None``.""" raise NotImplementedError @abstractmethod async def list_members(self, position_id: str) -> list[dict[str, Any]]: """Return ``[{id, email}, ...]`` of users assigned to this position.""" raise NotImplementedError @abstractmethod async def assign_users(self, position_id: str, user_ids: list[str]) -> int: """Assign a position to many users (1:1 — overwrites any prior position). Returns the number of users updated. """ raise NotImplementedError @abstractmethod async def unassign_users(self, user_ids: list[str]) -> int: """Clear the position of many users. Returns the number updated.""" raise NotImplementedError @abstractmethod async def list_member_ids(self, position_id: str) -> list[str]: """Return just the user ids assigned to this position (sync fan-out).""" raise NotImplementedError @abstractmethod async def list_assigned_user_ids(self) -> list[str]: """Return every user id that has an explicit position (for bulk resync).""" raise NotImplementedError @abstractmethod async def list_all_users(self) -> list[dict[str, Any]]: """Return ``[{id, email, system_role, position_id}, ...]`` for every user. Powers the admin member-picker and the per-user position editor. """ raise NotImplementedError