113 lines
3.5 KiB
Python
113 lines
3.5 KiB
Python
|
"""Schemas for Forgejo connection CRUD API payloads."""
|
||
|
|
|
||
|
|
from __future__ import annotations
|
||
|
|
|
||
|
|
from datetime import datetime
|
||
|
|
from uuid import UUID
|
||
|
|
|
||
|
|
from pydantic import field_validator
|
||
|
|
from sqlmodel import Field, SQLModel
|
||
|
|
|
||
|
|
RUNTIME_ANNOTATION_TYPES = (datetime, UUID)
|
||
|
|
|
||
|
|
|
||
|
|
class ForgejoConnectionBase(SQLModel):
|
||
|
|
"""Shared connection fields used across create/read payloads."""
|
||
|
|
|
||
|
|
name: str
|
||
|
|
base_url: str
|
||
|
|
token: str | None = None
|
||
|
|
active: bool = True
|
||
|
|
|
||
|
|
@field_validator("base_url", mode="before")
|
||
|
|
@classmethod
|
||
|
|
def normalize_base_url(cls, value: object) -> str | None | object:
|
||
|
|
"""Normalize base_url - ensure it's a valid http/https URL without /api/v1 path."""
|
||
|
|
if value is None:
|
||
|
|
return None
|
||
|
|
if isinstance(value, str):
|
||
|
|
value = value.strip()
|
||
|
|
if not value:
|
||
|
|
return None
|
||
|
|
# Remove trailing slashes
|
||
|
|
value = value.rstrip("/")
|
||
|
|
# Validate protocol
|
||
|
|
if not value.startswith(("http://", "https://")):
|
||
|
|
raise ValueError("base_url must be http:// or https://")
|
||
|
|
# Remove /api/v1 if present
|
||
|
|
if "/api/v1" in value:
|
||
|
|
# Find the base host
|
||
|
|
import re
|
||
|
|
match = re.match(r"(https?://[^/]+)", value)
|
||
|
|
if match:
|
||
|
|
value = match.group(1).rstrip("/")
|
||
|
|
return value
|
||
|
|
return value
|
||
|
|
|
||
|
|
@field_validator("token", mode="before")
|
||
|
|
@classmethod
|
||
|
|
def normalize_token(cls, value: object) -> str | None | object:
|
||
|
|
"""Normalize empty/whitespace tokens to `None`."""
|
||
|
|
if value is None:
|
||
|
|
return None
|
||
|
|
if isinstance(value, str):
|
||
|
|
value = value.strip()
|
||
|
|
return value or None
|
||
|
|
return value
|
||
|
|
|
||
|
|
|
||
|
|
class ForgejoConnectionCreate(ForgejoConnectionBase):
|
||
|
|
"""Payload for creating a Forgejo connection configuration."""
|
||
|
|
|
||
|
|
|
||
|
|
class ForgejoConnectionUpdate(SQLModel):
|
||
|
|
"""Payload for partial Forgejo connection updates."""
|
||
|
|
|
||
|
|
name: str | None = None
|
||
|
|
base_url: str | None = None
|
||
|
|
token: str | None = None
|
||
|
|
active: bool | None = None
|
||
|
|
|
||
|
|
@field_validator("base_url", mode="before")
|
||
|
|
@classmethod
|
||
|
|
def normalize_base_url(cls, value: object) -> str | None | object:
|
||
|
|
"""Normalize base_url - ensure it's a valid http/https URL without /api/v1 path."""
|
||
|
|
if value is None:
|
||
|
|
return None
|
||
|
|
if isinstance(value, str):
|
||
|
|
value = value.strip()
|
||
|
|
if not value:
|
||
|
|
return None
|
||
|
|
value = value.rstrip("/")
|
||
|
|
if not value.startswith(("http://", "https://")):
|
||
|
|
raise ValueError("base_url must be http:// or https://")
|
||
|
|
if "/api/v1" in value:
|
||
|
|
import re
|
||
|
|
match = re.match(r"(https?://[^/]+)", value)
|
||
|
|
if match:
|
||
|
|
value = match.group(1).rstrip("/")
|
||
|
|
return value
|
||
|
|
return value
|
||
|
|
|
||
|
|
@field_validator("token", mode="before")
|
||
|
|
@classmethod
|
||
|
|
def normalize_token(cls, value: object) -> str | None | object:
|
||
|
|
"""Normalize empty/whitespace tokens to `None`."""
|
||
|
|
if value is None:
|
||
|
|
return None
|
||
|
|
if isinstance(value, str):
|
||
|
|
value = value.strip()
|
||
|
|
return value or None
|
||
|
|
return value
|
||
|
|
|
||
|
|
|
||
|
|
class ForgejoConnectionRead(ForgejoConnectionBase):
|
||
|
|
"""Connection payload returned from read endpoints."""
|
||
|
|
|
||
|
|
id: UUID
|
||
|
|
organization_id: UUID
|
||
|
|
has_token: bool
|
||
|
|
token_last_eight: str | None
|
||
|
|
created_at: datetime
|
||
|
|
updated_at: datetime
|