246 lines
7.3 KiB
Python
246 lines
7.3 KiB
Python
from datetime import datetime
|
|
from uuid import UUID
|
|
from zoneinfo import ZoneInfo, ZoneInfoNotFoundError
|
|
|
|
from pydantic import BaseModel, ConfigDict, Field, field_validator, model_validator
|
|
|
|
|
|
class InitializeRequest(BaseModel):
|
|
username: str = Field(min_length=3, max_length=64, pattern=r"^[A-Za-z0-9_.-]+$")
|
|
password: str = Field(min_length=12, max_length=256)
|
|
|
|
|
|
class LoginRequest(BaseModel):
|
|
username: str
|
|
password: str
|
|
|
|
|
|
class ChangePasswordRequest(BaseModel):
|
|
current_password: str = Field(min_length=1, max_length=256)
|
|
new_password: str = Field(min_length=12, max_length=256)
|
|
|
|
@model_validator(mode="after")
|
|
def reject_same_password(self):
|
|
if self.current_password == self.new_password:
|
|
raise ValueError("新密码不能与当前密码相同")
|
|
return self
|
|
|
|
|
|
class UserOut(BaseModel):
|
|
model_config = ConfigDict(from_attributes=True)
|
|
id: UUID
|
|
username: str
|
|
timezone: str
|
|
|
|
|
|
class UserUpdate(BaseModel):
|
|
timezone: str = Field(min_length=1, max_length=64)
|
|
|
|
@field_validator("timezone")
|
|
@classmethod
|
|
def validate_timezone(cls, value: str) -> str:
|
|
try:
|
|
ZoneInfo(value)
|
|
except (ZoneInfoNotFoundError, ValueError) as exc:
|
|
raise ValueError("timezone must be a valid IANA timezone") from exc
|
|
return value
|
|
|
|
|
|
class SessionOut(BaseModel):
|
|
model_config = ConfigDict(from_attributes=True)
|
|
id: UUID
|
|
current: bool
|
|
created_at: datetime
|
|
last_seen_at: datetime
|
|
expires_at: datetime
|
|
ip_address: str | None
|
|
user_agent: str | None
|
|
|
|
|
|
class NameUpdate(BaseModel):
|
|
name: str = Field(min_length=1, max_length=120)
|
|
|
|
@field_validator("name")
|
|
@classmethod
|
|
def clean_name(cls, value):
|
|
value = value.strip()
|
|
if not value:
|
|
raise ValueError("name cannot be blank")
|
|
return value
|
|
|
|
|
|
class FolderCreate(NameUpdate):
|
|
pass
|
|
|
|
|
|
class FolderOut(BaseModel):
|
|
model_config = ConfigDict(from_attributes=True)
|
|
id: UUID
|
|
name: str
|
|
|
|
|
|
class ListCreate(NameUpdate):
|
|
folder_id: UUID | None = None
|
|
|
|
|
|
class ListOut(BaseModel):
|
|
model_config = ConfigDict(from_attributes=True)
|
|
id: UUID
|
|
folder_id: UUID | None
|
|
name: str
|
|
is_inbox: bool
|
|
position: int
|
|
|
|
|
|
class ListMove(BaseModel):
|
|
folder_id: UUID | None
|
|
list_ids: list[UUID] = Field(min_length=1)
|
|
|
|
@model_validator(mode="after")
|
|
def unique_ids(self):
|
|
if len(self.list_ids) != len(set(self.list_ids)):
|
|
raise ValueError("list_ids must be unique")
|
|
return self
|
|
|
|
|
|
class ListReorder(BaseModel):
|
|
folder_id: UUID | None
|
|
list_ids: list[UUID] = Field(min_length=1)
|
|
|
|
@model_validator(mode="after")
|
|
def unique_ids(self):
|
|
if len(self.list_ids) != len(set(self.list_ids)):
|
|
raise ValueError("list_ids must be unique")
|
|
return self
|
|
|
|
|
|
class TaskCreate(BaseModel):
|
|
title: str = Field(min_length=1, max_length=500)
|
|
list_id: UUID
|
|
description: str = ""
|
|
priority: int = Field(default=0, ge=0, le=3)
|
|
due_at: datetime | None = None
|
|
due_has_time: bool = False
|
|
parent_id: UUID | None = None
|
|
rrule: str | None = Field(default=None, min_length=5, max_length=1000)
|
|
trigger_mode: str | None = Field(default=None, pattern="^(scheduled|after_completion)$")
|
|
after_completion_days: int | None = Field(default=None, ge=1, le=3650)
|
|
|
|
@field_validator("title")
|
|
@classmethod
|
|
def clean_title(cls, value):
|
|
value = value.strip()
|
|
if not value:
|
|
raise ValueError("title cannot be blank")
|
|
return value
|
|
|
|
@model_validator(mode="after")
|
|
def validate_recurrence(self):
|
|
has_recurrence = self.rrule is not None or self.trigger_mode is not None or self.after_completion_days is not None
|
|
if has_recurrence and self.due_at is None:
|
|
raise ValueError("recurrence requires due_at")
|
|
if has_recurrence and self.parent_id is not None:
|
|
raise ValueError("only top-level tasks can recur")
|
|
if self.trigger_mode == "after_completion":
|
|
if self.after_completion_days is None or self.rrule is not None:
|
|
raise ValueError("after_completion requires days and no rrule")
|
|
elif self.trigger_mode == "scheduled" and self.rrule is None:
|
|
raise ValueError("scheduled recurrence requires rrule")
|
|
elif self.trigger_mode is None and self.after_completion_days is not None:
|
|
raise ValueError("after_completion_days requires after_completion mode")
|
|
return self
|
|
|
|
|
|
class TaskUpdate(BaseModel):
|
|
title: str | None = Field(default=None, min_length=1, max_length=500)
|
|
description: str | None = None
|
|
priority: int | None = Field(default=None, ge=0, le=3)
|
|
due_at: datetime | None = None
|
|
due_has_time: bool | None = None
|
|
completed: bool | None = None
|
|
list_id: UUID | None = None
|
|
version: int = Field(ge=1)
|
|
|
|
@field_validator("title")
|
|
@classmethod
|
|
def clean_title(cls, value):
|
|
if value is None:
|
|
return value
|
|
value = value.strip()
|
|
if not value:
|
|
raise ValueError("title cannot be blank")
|
|
return value
|
|
|
|
@model_validator(mode="after")
|
|
def reject_null_non_nullable_fields(self):
|
|
for field in ("title", "description", "priority", "completed", "list_id", "due_has_time"):
|
|
if field in self.model_fields_set and getattr(self, field) is None:
|
|
raise ValueError(f"{field} cannot be null")
|
|
return self
|
|
|
|
|
|
class TaskReorder(BaseModel):
|
|
task_ids: list[UUID] = Field(min_length=1)
|
|
|
|
@model_validator(mode="after")
|
|
def unique_ids(self):
|
|
if len(self.task_ids) != len(set(self.task_ids)):
|
|
raise ValueError("task_ids must be unique")
|
|
return self
|
|
|
|
|
|
class TaskOut(BaseModel):
|
|
model_config = ConfigDict(from_attributes=True)
|
|
id: UUID
|
|
list_id: UUID
|
|
parent_id: UUID | None
|
|
title: str
|
|
description: str
|
|
priority: int
|
|
completed: bool
|
|
completed_at: datetime | None
|
|
due_at: datetime | None
|
|
due_has_time: bool
|
|
version: int
|
|
|
|
|
|
class TaskDetailOut(TaskOut):
|
|
subtasks: list[TaskOut] = Field(default_factory=list)
|
|
|
|
|
|
class TaskPage(BaseModel):
|
|
items: list[TaskDetailOut]
|
|
next_cursor: str | None = None
|
|
total: int = 0
|
|
page: int = 1
|
|
page_size: int = 50
|
|
|
|
|
|
class BatchTaskUpdate(BaseModel):
|
|
task_ids: list[UUID] = Field(min_length=1)
|
|
completed: bool | None = None
|
|
list_id: UUID | None = None
|
|
due_at: datetime | None = None
|
|
soft_delete: bool | None = None
|
|
versions: dict[UUID, int] | None = None
|
|
|
|
@model_validator(mode="after")
|
|
def require_operation(self):
|
|
operations = {"completed", "list_id", "due_at", "soft_delete"}
|
|
if not self.model_fields_set.intersection(operations):
|
|
raise ValueError("at least one batch operation is required")
|
|
if self.soft_delete is False:
|
|
raise ValueError("soft_delete can only be true")
|
|
if self.completed is True and self.versions is None:
|
|
raise ValueError("completion requires versions")
|
|
if self.versions is not None:
|
|
if set(self.versions) != set(self.task_ids):
|
|
raise ValueError("versions must cover every task")
|
|
if any(version < 1 for version in self.versions.values()):
|
|
raise ValueError("versions must be positive")
|
|
return self
|
|
|
|
|
|
class BatchResult(BaseModel):
|
|
updated: int
|