143 lines
4.2 KiB
Python
143 lines
4.2 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 ForgejoConnectionTestRequest(SQLModel):
|
|
"""Payload for testing a connection before saving it."""
|
|
|
|
base_url: str
|
|
token: str
|
|
|
|
|
|
class ForgejoConnectionTestRepo(SQLModel):
|
|
"""Minimal repo info returned in a connection test."""
|
|
|
|
full_name: str
|
|
name: str
|
|
owner: str
|
|
default_branch: str
|
|
private: bool
|
|
description: str | None = None
|
|
|
|
|
|
class ForgejoConnectionTestResponse(SQLModel):
|
|
"""Result of a pre-save connection test."""
|
|
|
|
valid: bool
|
|
user_login: str | None = None
|
|
user_full_name: str | None = None
|
|
repo_count: int = 0
|
|
repos: list[ForgejoConnectionTestRepo] = []
|
|
error: str | None = None
|
|
response_time_ms: float = 0.0
|
|
|
|
|
|
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
|