core_schemas.py

"""Pydantic v2 models for the Access API.

Each Pydantic class is the canonical shape for a specific endpoint surface;
there is no per-call field projection.

Polymorphic groups are modelled as discriminated unions on the `type` field.
Two union shapes are exposed here:

  - `GroupDetail`  — full detail (used by `/api/groups/{id}`)
  - `GroupSummary` — compact list view (used by `/api/groups`)

Request-body unions for create/update live in `requests_schemas.py`.

To keep field counts manageable, deeply nested membership/role/tag
relationships are emitted via dedicated child schemas, not full polymorphic
re-entries.
"""

from __future__ import annotations

from typing import Annotated, Any, Literal, Optional, Union

from pydantic import BaseModel, ConfigDict, Field, field_validator
from typing_extensions import TypeAliasType

from api.schemas.datetimes import FlexibleDatetime


# --- Tags -------------------------------------------------------------------


class TagDetail(BaseModel):
    model_config = ConfigDict(from_attributes=True)
    id: str
    name: str
    description: Optional[str] = None
    constraints: dict[str, Any] = Field(default_factory=dict)
    enabled: bool = True
    created_at: FlexibleDatetime
    updated_at: FlexibleDatetime
    deleted_at: Optional[FlexibleDatetime] = None
    # Resolved post-class via model_rebuild() — OktaGroupTagMapDetail is defined
    # below.
    active_group_tags: list["OktaGroupTagMapDetail"] = Field(default_factory=list)
    # Tag detail also lists the apps this tag is attached to. Flask
    # `TagResource.get()` `exclude=("all_group_tags", "all_app_tags")`
    # retains the `active_app_tags` projection.
    active_app_tags: list["AppTagMapDetail"] = Field(default_factory=list)


class TagSummary(BaseModel):
    model_config = ConfigDict(from_attributes=True)
    id: str
    name: str
    constraints: dict[str, Any] = Field(default_factory=dict)
    enabled: bool = True


class TagListItem(BaseModel):
    """Tag list-endpoint item. Slim field set (id, name, description,
    enabled, constraints, created_at, updated_at) — does not hydrate
    `active_group_tags`, which would be an N+1 across the page."""

    model_config = ConfigDict(from_attributes=True)
    id: str
    name: str
    description: Optional[str] = None
    constraints: dict[str, Any] = Field(default_factory=dict)
    enabled: bool = True
    created_at: FlexibleDatetime
    updated_at: FlexibleDatetime


class OktaGroupTagMapDetail(BaseModel):
    model_config = ConfigDict(from_attributes=True)
    created_at: FlexibleDatetime
    ended_at: Optional[FlexibleDatetime] = None
    active_tag: Optional[TagSummary] = None
    # Populated when the row is reached from the Tag side (`tag.active_group_tags`):
    active_group: Optional["_GroupRefForMembership"] = None
    active_app_tag_mapping: Optional["AppTagMapDetail"] = None


class AppTagMapDetail(BaseModel):
    model_config = ConfigDict(from_attributes=True)
    created_at: FlexibleDatetime
    ended_at: Optional[FlexibleDatetime] = None
    active_tag: Optional[TagSummary] = None
    # Populated when the row is reached from the Tag side
    # (`tag.active_app_tags`). Flask emitted `active_app.{id, name, description}`
    # via the legacy AppTagMapSchema only-list.
    active_app: Optional["AppSummary"] = None


# --- Apps -------------------------------------------------------------------


class AppIdRef(BaseModel):
    """Inline reference to an App by id (used in compact group views).

    Exposes the lifecycle plugin id on every embedded app reference so the
    React frontend can dispatch on plugin behaviour without a follow-up
    `/api/apps/{id}` fetch.
    """

    model_config = ConfigDict(from_attributes=True)
    id: str
    name: Optional[str] = None
    deleted_at: Optional[FlexibleDatetime] = None
    app_group_lifecycle_plugin: Optional[str] = None


class AppSummary(BaseModel):
    model_config = ConfigDict(from_attributes=True)
    id: str
    name: str
    description: Optional[str] = None
    created_at: FlexibleDatetime
    updated_at: FlexibleDatetime
    deleted_at: Optional[FlexibleDatetime] = None


class AppDetail(AppSummary):
    """Full App detail.

    The app's groups are intentionally NOT inlined here. An app can own
    hundreds of groups, and inlining each group's full membership made a
    single detail response materialize thousands of member rows. Groups (with
    their members) are served by the paginated `GET /api/apps/{id}/groups`
    endpoint (`AppGroupForAppDetail`) so the cost is bounded per page.
    """

    app_group_lifecycle_plugin: Optional[str] = None
    plugin_data: Optional[dict[str, Any]] = None
    active_app_tags: list[AppTagMapDetail] = Field(default_factory=list)


# --- Users ------------------------------------------------------------------


class OktaUserSummary(BaseModel):
    model_config = ConfigDict(from_attributes=True)
    id: str
    email: str
    first_name: Optional[str] = None
    last_name: Optional[str] = None
    display_name: Optional[str] = None
    # The list endpoint includes timestamps so the frontend can sort/group by
    # creation time. Flask exposed these via Marshmallow's `only=(...)` tuple;
    # leaving them off the Pydantic summary loses that capability.
    created_at: Optional[FlexibleDatetime] = None
    updated_at: Optional[FlexibleDatetime] = None
    deleted_at: Optional[FlexibleDatetime] = None


def _filter_profile_attrs(value: Any) -> dict[str, Any]:
    """Filter `OktaUser.profile` to keys in `USER_DISPLAY_CUSTOM_ATTRIBUTES`.

    Marshmallow applied the same filter via `OktaUserSchema.get_attribute`;
    the FastAPI side reuses this helper anywhere `profile` is exposed
    (top-level on `OktaUserDetail`, nested on `OktaUserManagerRef`)."""
    from api.config import settings

    attrs_to_display = [a for a in settings.USER_DISPLAY_CUSTOM_ATTRIBUTES.split(",") if a]
    if not attrs_to_display:
        return {}
    if not isinstance(value, dict):
        return {}
    return {key: value.get(key) for key in attrs_to_display}


class OktaUserManagerRef(OktaUserSummary):
    """Embedded manager reference inside `OktaUserDetail`.

    Retains `profile` (filtered to `USER_DISPLAY_CUSTOM_ATTRIBUTES`) so the
    React user-detail page can read `manager.profile.Title` to render the
    manager's job title alongside their name."""

    profile: dict[str, Any] = Field(default_factory=dict)

    @field_validator("profile", mode="before")
    @classmethod
    def _filter_profile(cls, value: Any) -> dict[str, Any]:
        return _filter_profile_attrs(value)


class OktaUserDetail(OktaUserSummary):
    profile: dict[str, Any] = Field(default_factory=dict)
    manager: Optional[OktaUserManagerRef] = None
    # Membership / ownership lists. The aggregated
    # `*_group_memberships_and_ownerships` pair Flask's `UserResource.get()`
    # excluded — they duplicate the data already in `active_group_memberships`
    # and `active_group_ownerships` and bloat the response. Resolved
    # post-class via model_rebuild() because OktaUserGroupMemberDetail is
    # defined further down.
    active_group_memberships: list["OktaUserGroupMemberDetail"] = Field(default_factory=list)
    active_group_ownerships: list["OktaUserGroupMemberDetail"] = Field(default_factory=list)

    @field_validator("profile", mode="before")
    @classmethod
    def _filter_profile(cls, value: Any) -> dict[str, Any]:
        return _filter_profile_attrs(value)


# --- Group memberships ------------------------------------------------------


class _GroupRefForMembership(BaseModel):
    """Compact group reference embedded in user/group membership rows."""

    model_config = ConfigDict(from_attributes=True)
    id: str
    type: str
    name: str
    description: Optional[str] = None
    is_owner: Optional[bool] = None
    is_managed: Optional[bool] = None
    deleted_at: Optional[FlexibleDatetime] = None
    app: Optional[AppIdRef] = None


class _RoleGroupMembershipRef(BaseModel):
    model_config = ConfigDict(from_attributes=True)
    id: str
    type: str
    name: str
    is_managed: Optional[bool] = None
    deleted_at: Optional[FlexibleDatetime] = None


class _RoleGroupMappingForMembership(BaseModel):
    model_config = ConfigDict(from_attributes=True)
    created_at: FlexibleDatetime
    ended_at: Optional[FlexibleDatetime] = None
    active_role_group: Optional[_RoleGroupMembershipRef] = None
    role_group: Optional[_RoleGroupMembershipRef] = None


class OktaUserGroupMemberDetail(BaseModel):
    model_config = ConfigDict(from_attributes=True)
    # `id`, `created_actor`, and `ended_actor` are read by the React renewal /
    # audit views (e.g. `BulkRenewal.tsx` uses `id` as a row key and
    # `created_actor` for the "added by" column). The actor relationships are
    # `lazy="raise_on_sql"`, so any route emitting this schema must eager-load
    # them — see `user_group_member_options` in `api/routers/_eager.py`.
    id: int
    is_owner: Optional[bool] = None
    created_at: FlexibleDatetime
    updated_at: Optional[FlexibleDatetime] = None
    ended_at: Optional[FlexibleDatetime] = None
    created_reason: Optional[str] = ""
    should_expire: Optional[bool] = None
    created_actor: Optional[OktaUserSummary] = None
    ended_actor: Optional[OktaUserSummary] = None
    user: Optional[OktaUserSummary] = None
    active_user: Optional[OktaUserSummary] = None
    group: Optional[_GroupRefForMembership] = None
    active_group: Optional[_GroupRefForMembership] = None
    role_group_mapping: Optional[_RoleGroupMappingForMembership] = None
    active_role_group_mapping: Optional[_RoleGroupMappingForMembership] = None


# --- Role group mappings ----------------------------------------------------


class RoleGroupMapDetail(BaseModel):
    model_config = ConfigDict(from_attributes=True)
    # `id`, `should_expire`, `created_reason`, and the actor relationships are
    # read by the role renewal / audit views (`BulkRenewal.tsx`). The actor
    # relationships are `lazy="raise_on_sql"` and must be eager-loaded — see
    # `role_group_map_options` in `api/routers/_eager.py`.
    id: int
    is_owner: Optional[bool] = None
    created_at: FlexibleDatetime
    ended_at: Optional[FlexibleDatetime] = None
    created_reason: Optional[str] = ""
    should_expire: Optional[bool] = None
    created_actor: Optional[OktaUserSummary] = None
    ended_actor: Optional[OktaUserSummary] = None
    role_group: Optional[_RoleGroupMembershipRef] = None
    active_role_group: Optional[_RoleGroupMembershipRef] = None
    group: Optional[_GroupRefForMembership] = None
    active_group: Optional[_GroupRefForMembership] = None


# --- Polymorphic groups -----------------------------------------------------
# Two discriminated unions: detail (Out) and summary (list view). The
# request-body unions live in `requests_schemas.py`.


class _GroupBase(BaseModel):
    model_config = ConfigDict(from_attributes=True)
    id: str
    name: str
    description: Optional[str] = ""
    is_managed: bool = True
    externally_managed_data: Optional[dict[str, Any]] = None
    created_at: FlexibleDatetime
    updated_at: FlexibleDatetime
    deleted_at: Optional[FlexibleDatetime] = None
    # Members are intentionally NOT inlined: a group can have thousands, and
    # inlining them made the detail response (and, via the app endpoint, every
    # group of an app) materialize unbounded member rows. They are served by
    # the paginated `GET /api/groups/{id}/member-details` endpoint instead.
    active_group_tags: list[OktaGroupTagMapDetail] = Field(default_factory=list)


class OktaGroupDetail(_GroupBase):
    type: Literal["okta_group"] = "okta_group"
    active_role_member_mappings: list[RoleGroupMapDetail] = Field(default_factory=list)
    active_role_owner_mappings: list[RoleGroupMapDetail] = Field(default_factory=list)


class RoleGroupDetail(_GroupBase):
    type: Literal["role_group"] = "role_group"
    active_role_associated_group_member_mappings: list[RoleGroupMapDetail] = Field(default_factory=list)
    active_role_associated_group_owner_mappings: list[RoleGroupMapDetail] = Field(default_factory=list)


class AppGroupDetail(_GroupBase):
    type: Literal["app_group"] = "app_group"
    app_id: Optional[str] = None
    is_owner: bool = False
    plugin_data: Optional[dict[str, Any]] = None
    app: Optional[AppIdRef] = None
    active_role_member_mappings: list[RoleGroupMapDetail] = Field(default_factory=list)
    active_role_owner_mappings: list[RoleGroupMapDetail] = Field(default_factory=list)


class AppGroupForAppDetail(BaseModel):
    """Slimmer shape used by the paginated `GET /api/apps/{id}/groups` endpoint.

    Members are NOT inlined: a single group can have thousands, and inlining
    them made even a 10-group page able to ship megabytes. Instead each item
    carries `member_count` / `owner_count` (cheap SQL aggregates); the UI fetches
    a group's members on demand from the paginated
    `GET /api/groups/{id}/member-details` endpoint."""

    model_config = ConfigDict(from_attributes=True)
    id: str
    type: Literal["app_group"] = "app_group"
    name: str
    description: Optional[str] = ""
    is_managed: bool = True
    externally_managed_data: Optional[dict[str, Any]] = None
    created_at: FlexibleDatetime
    updated_at: FlexibleDatetime
    deleted_at: Optional[FlexibleDatetime] = None
    app_id: Optional[str] = None
    is_owner: bool = False
    plugin_data: Optional[dict[str, Any]] = None
    app: Optional[AppIdRef] = None
    member_count: int = 0
    owner_count: int = 0


# Named via `TypeAliasType` so FastAPI emits a single `GroupDetail` schema
# (a `$ref`) instead of an inline `anyOf`; this gives the OpenAPI→TypeScript
# codegen a clean reusable union type. Purely a naming hint — validation and
# wire shape are identical to the bare `Annotated[Union[...]]`.
GroupDetail = TypeAliasType(
    "GroupDetail",
    Annotated[
        Union[OktaGroupDetail, RoleGroupDetail, AppGroupDetail],
        Field(discriminator="type"),
    ],
)


# --- Group summaries (list endpoints) --------------------------------------


class _GroupSummaryBase(BaseModel):
    model_config = ConfigDict(from_attributes=True)
    id: str
    name: str
    description: Optional[str] = ""
    is_managed: bool = True
    created_at: FlexibleDatetime
    updated_at: FlexibleDatetime
    active_group_tags: list[OktaGroupTagMapDetail] = Field(default_factory=list)


class OktaGroupSummary(_GroupSummaryBase):
    type: Literal["okta_group"] = "okta_group"


class RoleGroupSummary(_GroupSummaryBase):
    type: Literal["role_group"] = "role_group"
    active_role_associated_group_member_mappings: list[RoleGroupMapDetail] = Field(default_factory=list)
    active_role_associated_group_owner_mappings: list[RoleGroupMapDetail] = Field(default_factory=list)


class AppGroupSummary(_GroupSummaryBase):
    type: Literal["app_group"] = "app_group"
    app: Optional[AppIdRef] = None


GroupSummary = TypeAliasType(
    "GroupSummary",
    Annotated[
        Union[OktaGroupSummary, RoleGroupSummary, AppGroupSummary],
        Field(discriminator="type"),
    ],
)


class RoleGroupListItem(BaseModel):
    """Slim row shape for `GET /api/roles` (id, type, name, description,
    created_at, updated_at).

    The role-list page does not render tags or role associations, so we pay
    neither the loader cost nor the JSON bloat from emitting them on every
    row."""

    model_config = ConfigDict(from_attributes=True)
    id: str
    type: Literal["role_group"] = "role_group"
    name: str
    description: Optional[str] = ""
    created_at: FlexibleDatetime
    updated_at: FlexibleDatetime


# --- Group references (embedded inside requests, audit, etc.) -------------
# A minimal polymorphic group shape that does NOT touch any "active_*" or
# "all_*" relationship attributes — those are `lazy="raise_on_sql"` and not
# always pre-loaded when groups are embedded inside other objects.


class _GroupRefBase(BaseModel):
    model_config = ConfigDict(from_attributes=True)
    id: str
    name: str
    description: Optional[str] = ""
    is_managed: bool = True
    deleted_at: Optional[FlexibleDatetime] = None


class OktaGroupRef(_GroupRefBase):
    type: Literal["okta_group"] = "okta_group"


class RoleGroupRef(_GroupRefBase):
    type: Literal["role_group"] = "role_group"


class AppGroupRef(_GroupRefBase):
    type: Literal["app_group"] = "app_group"
    app_id: Optional[str] = None
    is_owner: bool = False
    app: Optional[AppIdRef] = None


GroupRef = TypeAliasType(
    "GroupRef",
    Annotated[
        Union[OktaGroupRef, RoleGroupRef, AppGroupRef],
        Field(discriminator="type"),
    ],
)


# Wider polymorphic refs surfaced inside `RoleRequestDetail`. The role-request
# detail page renders the role's current members and the target group's tags
# inline, so those two relationship arrays must travel with the embedded
# group refs (the bare `GroupRef` shape intentionally omits them to keep
# audit/list payloads slim).
class RoleRequestRequesterRoleRef(_GroupRefBase):
    type: Literal["role_group"] = "role_group"
    active_user_memberships: list[OktaUserGroupMemberDetail] = Field(default_factory=list)


class RoleRequestRequestedOktaGroupRef(_GroupRefBase):
    type: Literal["okta_group"] = "okta_group"
    active_group_tags: list[OktaGroupTagMapDetail] = Field(default_factory=list)


class RoleRequestRequestedAppGroupRef(_GroupRefBase):
    type: Literal["app_group"] = "app_group"
    app_id: Optional[str] = None
    is_owner: bool = False
    app: Optional[AppIdRef] = None
    active_group_tags: list[OktaGroupTagMapDetail] = Field(default_factory=list)


RoleRequestRequestedGroupRef = TypeAliasType(
    "RoleRequestRequestedGroupRef",
    Annotated[
        Union[RoleRequestRequestedOktaGroupRef, RoleRequestRequestedAppGroupRef],
        Field(discriminator="type"),
    ],
)


# --- Members views (list-of-IDs response shapes) ----------------------------


class GroupMembersSummary(BaseModel):
    """Wire shape for `GET/PUT /api/groups/{id}/members`."""

    members: list[str]
    owners: list[str]


class RoleMembersSummary(BaseModel):
    """Wire shape for `GET/PUT /api/roles/{id}/members`."""

    groups_in_role: list[str]
    groups_owned_by_role: list[str]


# --- Error envelope ---------------------------------------------------------


class ProblemDetailError(BaseModel):
    """One entry in the non-standard `errors` extension that validation
    problems carry (see `_validation_errors` in `api/exception_handlers.py`)."""

    type: Optional[str] = None
    loc: Optional[list[Union[str, int]]] = None
    msg: Optional[str] = None
    ctx: Optional[dict[str, Any]] = None


class ProblemDetail(BaseModel):
    """RFC 9457 problem-detail envelope emitted for every HTTP error by
    `api/exception_handlers.py` (Content-Type `application/problem+json`).

    Declared here so routers can advertise it via `responses={...}`; the
    generated TypeScript client then types its `*Error` payloads against this
    shape. Mirrors the hand-written `ErrorMessage` type in
    `src/api/apiFetcher.ts`. Every field is optional (`type` carries the RFC
    default, the rest are populated per error)."""

    type: str = "about:blank"
    title: Optional[str] = None
    status: Optional[int] = None
    detail: Optional[str] = None
    errors: Optional[list[ProblemDetailError]] = None


# Manage forward refs after all classes are defined
OktaUserDetail.model_rebuild()
AppDetail.model_rebuild()
TagDetail.model_rebuild()
OktaGroupTagMapDetail.model_rebuild()
AppTagMapDetail.model_rebuild()