UMBRA/backend/app/schemas/connection.py
Kyle Pope 3d22568b9c Add user connections, notification centre, and people integration
Implements the full User Connections & Notification Centre feature:

Phase 1 - Database: migrations 039-043 adding umbral_name to users,
profile/social fields to settings, notifications table, connection
request/user_connection tables, and linked_user_id to people.

Phase 2 - Notifications: backend CRUD router + service + 90-day purge,
frontend NotificationsPage with All/Unread filter, bell icon in sidebar
with unread badge polling every 60s.

Phase 3 - Settings: profile fields (phone, mobile, address, company,
job_title), social card with accept_connections toggle and per-field
sharing defaults, umbral name display with CopyableField.

Phase 4 - Connections: timing-safe user search, send/accept/reject flow
with atomic status updates, bidirectional UserConnection + Person records,
in-app + ntfy notifications, per-receiver pending cap, nginx rate limiting.

Phase 5 - People integration: batch-loaded shared profiles (N+1 prevention),
Ghost icon for umbral contacts, Umbral filter pill, split Add Person button,
shared field indicators (synced labels + Lock icons), disabled form inputs
for synced fields on umbral contacts.

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
2026-03-04 02:10:16 +08:00

76 lines
2.1 KiB
Python

"""
Connection schemas — search, request, respond, connection management.
All input schemas use extra="forbid" to prevent mass-assignment.
"""
import re
from typing import Literal, Optional
from datetime import datetime
from pydantic import BaseModel, ConfigDict, Field, field_validator
_UMBRAL_NAME_RE = re.compile(r'^[a-zA-Z0-9_-]{3,50}$')
class UmbralSearchRequest(BaseModel):
model_config = ConfigDict(extra="forbid")
umbral_name: str = Field(..., max_length=50)
@field_validator('umbral_name')
@classmethod
def validate_umbral_name(cls, v: str) -> str:
if not _UMBRAL_NAME_RE.match(v):
raise ValueError('Umbral name must be 3-50 alphanumeric characters, hyphens, or underscores')
return v
class UmbralSearchResponse(BaseModel):
found: bool
class SendConnectionRequest(BaseModel):
model_config = ConfigDict(extra="forbid")
umbral_name: str = Field(..., max_length=50)
@field_validator('umbral_name')
@classmethod
def validate_umbral_name(cls, v: str) -> str:
if not _UMBRAL_NAME_RE.match(v):
raise ValueError('Umbral name must be 3-50 alphanumeric characters, hyphens, or underscores')
return v
class ConnectionRequestResponse(BaseModel):
id: int
sender_umbral_name: str
sender_preferred_name: Optional[str] = None
receiver_umbral_name: str
receiver_preferred_name: Optional[str] = None
status: str
created_at: datetime
class RespondRequest(BaseModel):
model_config = ConfigDict(extra="forbid")
action: Literal["accept", "reject"]
class ConnectionResponse(BaseModel):
id: int
connected_user_id: int
connected_umbral_name: str
connected_preferred_name: Optional[str] = None
person_id: Optional[int] = None
created_at: datetime
class SharingOverrideUpdate(BaseModel):
model_config = ConfigDict(extra="forbid")
preferred_name: Optional[bool] = None
email: Optional[bool] = None
phone: Optional[bool] = None
mobile: Optional[bool] = None
birthday: Optional[bool] = None
address: Optional[bool] = None
company: Optional[bool] = None
job_title: Optional[bool] = None