| from datetime import datetime |
| from enum import StrEnum |
| from typing import Any |
|
|
| from pydantic import BaseModel, ConfigDict, Field, field_validator |
|
|
|
|
| class UserProvider(StrEnum): |
| LOCAL = "local" |
| GOOGLE = "google" |
| LOCAL_GOOGLE = "local_google" |
|
|
|
|
| class PlatformRole(StrEnum): |
| USER = "user" |
| PLATFORM_ADMIN = "platform_admin" |
|
|
|
|
| class OrganizationRole(StrEnum): |
| OWNER = "owner" |
| |
| LEGACY_ADMIN = "admin" |
| MEMBER = "member" |
| VIEWER = "viewer" |
|
|
|
|
| class SourceScope(StrEnum): |
| PUBLIC_VBPL = "public_vbpl" |
| GUEST_SESSION = "guest_session" |
| USER_PRIVATE = "user_private" |
| ORG_PRIVATE = "org_private" |
|
|
|
|
| class AuthorityLevel(StrEnum): |
| PUBLIC_LAW = "public_law" |
| INTERNAL_POLICY = "internal_policy" |
| UPLOADED_CONTEXT = "uploaded_context" |
|
|
|
|
| class SourceDocumentStatus(StrEnum): |
| ACTIVE = "active" |
| OUTDATED = "outdated" |
| DRAFT = "draft" |
| UNKNOWN = "unknown" |
| PENDING_INDEX = "pending_index" |
| INDEXING_FAILED = "indexing_failed" |
|
|
|
|
| class TokenType(StrEnum): |
| ACCESS = "access" |
| REFRESH = "refresh" |
| OAUTH_STATE = "oauth_state" |
|
|
|
|
| class UserPublic(BaseModel): |
| id: str |
| email: str |
| full_name: str | None = None |
| avatar_url: str | None = None |
| provider: UserProvider |
| |
| platform_role: PlatformRole |
| is_active: bool |
| is_verified: bool |
| created_at: datetime |
| updated_at: datetime |
|
|
| model_config = ConfigDict(from_attributes=True) |
|
|
|
|
| class MembershipPublic(BaseModel): |
| id: str |
| organization_id: str |
| organization_name: str |
| organization_slug: str |
| role: OrganizationRole |
| created_at: datetime |
| updated_at: datetime |
|
|
| model_config = ConfigDict(from_attributes=True) |
|
|
|
|
| class CurrentUserResponse(BaseModel): |
| user: UserPublic |
| memberships: list[MembershipPublic] = Field(default_factory=list) |
|
|
|
|
| class RegisterRequest(BaseModel): |
| email: str = Field(min_length=3, max_length=320) |
| password: str = Field(min_length=8, max_length=200) |
| full_name: str | None = Field(default=None, max_length=200) |
|
|
| model_config = ConfigDict(extra="forbid") |
|
|
| @field_validator("email") |
| @classmethod |
| def normalize_email(cls, value: str) -> str: |
| email = value.strip().lower() |
| if "@" not in email or email.startswith("@") or email.endswith("@"): |
| raise ValueError("valid email is required") |
| return email |
|
|
|
|
| class LoginRequest(BaseModel): |
| email: str = Field(min_length=3, max_length=320) |
| password: str = Field(min_length=1, max_length=200) |
|
|
| @field_validator("email") |
| @classmethod |
| def normalize_email(cls, value: str) -> str: |
| return value.strip().lower() |
|
|
|
|
| class RefreshTokenRequest(BaseModel): |
| refresh_token: str = Field(min_length=16) |
|
|
|
|
| class LogoutRequest(BaseModel): |
| refresh_token: str = Field(min_length=16) |
|
|
|
|
| class TokenResponse(BaseModel): |
| access_token: str |
| refresh_token: str |
| token_type: str = "bearer" |
| access_token_expires_in: int |
| refresh_token_expires_in: int |
|
|
|
|
| class OrganizationCreateRequest(BaseModel): |
| name: str = Field(min_length=1, max_length=200) |
| slug: str | None = Field(default=None, min_length=1, max_length=120) |
|
|
|
|
| class OrganizationPublic(BaseModel): |
| id: str |
| name: str |
| slug: str |
| created_by_user_id: str |
| created_at: datetime |
| updated_at: datetime |
|
|
| model_config = ConfigDict(from_attributes=True) |
|
|
|
|
| class OrganizationCreateResponse(BaseModel): |
| organization: OrganizationPublic |
| membership: MembershipPublic |
| invite_code: str |
|
|
|
|
| class OrganizationListResponse(BaseModel): |
| organizations: list[OrganizationPublic] |
| memberships: list[MembershipPublic] |
| invite_code: str | None = None |
|
|
|
|
| class OrganizationJoinRequest(BaseModel): |
| invite_code: str = Field(min_length=16, max_length=200) |
|
|
|
|
| class OrganizationJoinResponse(BaseModel): |
| organization: OrganizationPublic |
| membership: MembershipPublic |
|
|
|
|
| class InviteCodeResponse(BaseModel): |
| organization_id: str |
| invite_code: str |
|
|
|
|
| class OrganizationMemberPublic(BaseModel): |
| id: str |
| user_id: str |
| email: str |
| full_name: str | None = None |
| role: OrganizationRole |
| created_at: datetime |
| updated_at: datetime |
|
|
|
|
| class OrganizationMembersResponse(BaseModel): |
| organization_id: str |
| members: list[OrganizationMemberPublic] |
|
|
|
|
| class SourceAccessContext(BaseModel): |
| user_id: str | None = None |
| guest_session_id: str | None = None |
| organization_roles: dict[str, OrganizationRole] = Field(default_factory=dict) |
|
|
|
|
| class SourceAccessRequest(BaseModel): |
| source_scope: SourceScope |
| owner_user_id: str | None = None |
| guest_session_id: str | None = None |
| org_id: str | None = None |
| uploaded_by: str | None = None |
| authority_level: AuthorityLevel |
| document_status: SourceDocumentStatus = SourceDocumentStatus.UNKNOWN |
| source_uri: str | None = None |
| metadata: dict[str, Any] = Field(default_factory=dict) |
|
|
|
|
| class SourceAccessDecision(BaseModel): |
| allowed: bool |
| reason: str |
|
|