Spaces:
Running
Running
File size: 4,945 Bytes
c91c7db 3493993 1fed801 c91c7db | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 | from __future__ import annotations
from datetime import datetime
from typing import Literal
from pydantic import BaseModel, ConfigDict, Field, field_validator, model_validator
from app.security.scopes import ALL_SCOPES
Environment = Literal["live", "test"]
class APIKeyCreate(BaseModel):
name: str = Field(min_length=1, max_length=120)
environment: Environment = "live"
role: str | None = Field(default="viewer", min_length=1, max_length=64)
scopes: list[str] = Field(default_factory=list)
expires_at: datetime | None = None
expires_in_seconds: int | None = Field(default=None, ge=60, le=31_536_000)
notes: str | None = Field(default=None, max_length=4000)
requests_per_minute: int | None = Field(default=None, ge=1, le=1_000_000)
concurrent_jobs: int | None = Field(default=None, ge=1, le=10_000)
uploads_per_hour: int | None = Field(default=None, ge=1, le=1_000_000)
processing_bytes_per_day: int | None = Field(default=None, ge=1_048_576)
@field_validator("name")
@classmethod
def validate_name(cls, value: str) -> str:
normalized = value.strip()
if not normalized:
raise ValueError("API key name cannot be blank")
return normalized
@field_validator("scopes")
@classmethod
def validate_scopes(cls, values: list[str]) -> list[str]:
normalized = list(dict.fromkeys(value.strip().lower() for value in values))
unknown = set(normalized) - ALL_SCOPES
if unknown:
raise ValueError(f"Unsupported scopes: {sorted(unknown)}")
return normalized
@model_validator(mode="after")
def validate_expiration(self) -> APIKeyCreate:
if self.expires_at is not None and self.expires_in_seconds is not None:
raise ValueError("Use either expires_at or expires_in_seconds, not both")
return self
class APIKeyPatch(BaseModel):
name: str | None = Field(default=None, min_length=1, max_length=120)
role: str | None = Field(default=None, min_length=1, max_length=64)
scopes: list[str] | None = None
expires_at: datetime | None = None
clear_expiration: bool = False
notes: str | None = Field(default=None, max_length=4000)
requests_per_minute: int | None = Field(default=None, ge=1, le=1_000_000)
concurrent_jobs: int | None = Field(default=None, ge=1, le=10_000)
uploads_per_hour: int | None = Field(default=None, ge=1, le=1_000_000)
processing_bytes_per_day: int | None = Field(default=None, ge=1_048_576)
@field_validator("name")
@classmethod
def validate_name(cls, value: str | None) -> str | None:
if value is None:
return None
normalized = value.strip()
if not normalized:
raise ValueError("API key name cannot be blank")
return normalized
@field_validator("scopes")
@classmethod
def validate_scopes(cls, values: list[str] | None) -> list[str] | None:
if values is None:
return None
normalized = list(dict.fromkeys(value.strip().lower() for value in values))
unknown = set(normalized) - ALL_SCOPES
if unknown:
raise ValueError(f"Unsupported scopes: {sorted(unknown)}")
return normalized
@model_validator(mode="after")
def validate_expiration(self) -> APIKeyPatch:
if self.clear_expiration and "expires_at" in self.model_fields_set:
raise ValueError("Use either expires_at or clear_expiration, not both")
return self
class APIKeyRotate(BaseModel):
grace_period_seconds: int = Field(default=0, ge=0, le=86_400)
class APIKeyView(BaseModel):
model_config = ConfigDict(from_attributes=True)
id: str
name: str
key_prefix: str
environment: str
status: str
role: str | None
scopes: list[str]
created_at: datetime
last_used_at: datetime | None
expires_at: datetime | None
grace_expires_at: datetime | None
created_by: str | None
notes: str | None
rotated_from_id: str | None
requests_per_minute: int
concurrent_jobs: int
uploads_per_hour: int
processing_bytes_per_day: int
class APIKeyCreated(APIKeyView):
api_key: str
class APIKeyList(BaseModel):
items: list[APIKeyView]
total: int
class AuthContextView(BaseModel):
id: str
name: str
key_prefix: str
environment: str
role: str | None
scopes: list[str]
expires_at: datetime | None
workspace_id: str | None = None
user_id: str | None = None
membership_role: str | None = None
class AuditLogView(BaseModel):
model_config = ConfigDict(from_attributes=True)
id: str
request_id: str
api_key_id: str | None
key_name: str | None
ip_address: str | None
user_agent: str | None
endpoint: str
http_method: str
response_code: int
processing_time_ms: int
bytes_uploaded: int
bytes_downloaded: int
created_at: datetime
|