nathanb
d024bf7fa3
- ui has been made 'kinda better' (after making it worse for a while lol - ESP rfid readers are now supported [ill upload the code for them in another repo later] - admin system has been secured a bit better and seems to be working well
617 lines
17 KiB
Python
617 lines
17 KiB
Python
from pydantic import BaseModel, EmailStr, Field, ConfigDict, field_serializer, field_validator
|
|
from typing import Optional, Literal, Any
|
|
from datetime import datetime, date
|
|
from ..core.datetime import to_utc_naive, to_zulu_iso
|
|
from ..models.models import (
|
|
UserRole,
|
|
MembershipStatus,
|
|
PaymentStatus,
|
|
PaymentMethod,
|
|
EspReaderProvisioningStatus,
|
|
EspReaderType,
|
|
EspTapAction,
|
|
RfidWriteJobStatus,
|
|
)
|
|
|
|
|
|
class UTCBaseModel(BaseModel):
|
|
@field_validator("*", mode="before", check_fields=False)
|
|
@classmethod
|
|
def normalize_datetime_inputs(cls, value: Any) -> Any:
|
|
if isinstance(value, datetime):
|
|
return to_utc_naive(value)
|
|
return value
|
|
|
|
@field_validator("*", mode="after", check_fields=False)
|
|
@classmethod
|
|
def normalize_parsed_datetimes(cls, value: Any) -> Any:
|
|
if isinstance(value, datetime):
|
|
return to_utc_naive(value)
|
|
return value
|
|
|
|
@field_serializer("*", when_used="json", check_fields=False)
|
|
def serialize_datetime_outputs(self, value: Any) -> Any:
|
|
if isinstance(value, datetime):
|
|
return to_zulu_iso(value)
|
|
return value
|
|
|
|
|
|
# User Schemas
|
|
class UserBase(UTCBaseModel):
|
|
email: EmailStr
|
|
first_name: str = Field(..., min_length=1, max_length=100)
|
|
last_name: str = Field(..., min_length=1, max_length=100)
|
|
phone: Optional[str] = None
|
|
address: Optional[str] = None
|
|
|
|
|
|
class UserCreate(UserBase):
|
|
password: str = Field(..., min_length=8)
|
|
|
|
|
|
class UserUpdate(UTCBaseModel):
|
|
email: Optional[EmailStr] = None
|
|
first_name: Optional[str] = Field(None, min_length=1, max_length=100)
|
|
last_name: Optional[str] = Field(None, min_length=1, max_length=100)
|
|
phone: Optional[str] = None
|
|
address: Optional[str] = None
|
|
role: Optional[UserRole] = None
|
|
volunteer_level: Optional[str] = Field(None, max_length=50)
|
|
|
|
|
|
class UserResponse(UserBase):
|
|
model_config = ConfigDict(from_attributes=True)
|
|
|
|
id: int
|
|
role: UserRole
|
|
volunteer_level: Optional[str] = None
|
|
is_active: bool
|
|
created_at: datetime
|
|
last_login: Optional[datetime] = None
|
|
|
|
|
|
class UserInDB(UserResponse):
|
|
hashed_password: str
|
|
|
|
|
|
# Authentication Schemas
|
|
class Token(UTCBaseModel):
|
|
access_token: str
|
|
token_type: str = "bearer"
|
|
|
|
|
|
class TokenData(UTCBaseModel):
|
|
user_id: Optional[int] = None
|
|
|
|
|
|
class LoginRequest(UTCBaseModel):
|
|
email: EmailStr
|
|
password: str
|
|
|
|
|
|
# Password Reset Schemas
|
|
class ForgotPasswordRequest(UTCBaseModel):
|
|
email: EmailStr
|
|
|
|
|
|
class ResetPasswordRequest(UTCBaseModel):
|
|
token: str = Field(..., min_length=1)
|
|
new_password: str = Field(..., min_length=8)
|
|
|
|
|
|
class ChangePasswordRequest(UTCBaseModel):
|
|
current_password: str = Field(..., min_length=1)
|
|
new_password: str = Field(..., min_length=8)
|
|
|
|
|
|
# Membership Tier Schemas
|
|
class MembershipTierBase(UTCBaseModel):
|
|
name: str = Field(..., min_length=1, max_length=100)
|
|
description: Optional[str] = None
|
|
annual_fee: float = Field(..., ge=0)
|
|
benefits: Optional[str] = None
|
|
|
|
|
|
class MembershipTierCreate(MembershipTierBase):
|
|
pass
|
|
|
|
|
|
class MembershipTierUpdate(UTCBaseModel):
|
|
name: Optional[str] = Field(None, min_length=1, max_length=100)
|
|
description: Optional[str] = None
|
|
annual_fee: Optional[float] = Field(None, ge=0)
|
|
benefits: Optional[str] = None
|
|
is_active: Optional[bool] = None
|
|
|
|
|
|
class MembershipTierResponse(MembershipTierBase):
|
|
model_config = ConfigDict(from_attributes=True)
|
|
|
|
id: int
|
|
is_active: bool
|
|
created_at: datetime
|
|
|
|
|
|
# Membership Schemas
|
|
class MembershipBase(UTCBaseModel):
|
|
tier_id: int
|
|
auto_renew: bool = False
|
|
|
|
|
|
class MembershipCreate(MembershipBase):
|
|
start_date: date
|
|
end_date: date
|
|
|
|
|
|
class MembershipUpdate(UTCBaseModel):
|
|
tier_id: Optional[int] = None
|
|
status: Optional[MembershipStatus] = None
|
|
end_date: Optional[date] = None
|
|
auto_renew: Optional[bool] = None
|
|
|
|
|
|
class MembershipResponse(UTCBaseModel):
|
|
model_config = ConfigDict(from_attributes=True)
|
|
|
|
id: int
|
|
user_id: int
|
|
tier_id: int
|
|
status: MembershipStatus
|
|
start_date: date
|
|
end_date: date
|
|
auto_renew: bool
|
|
created_at: datetime
|
|
tier: MembershipTierResponse
|
|
|
|
|
|
# Payment Schemas
|
|
class PaymentBase(UTCBaseModel):
|
|
amount: float = Field(..., gt=0)
|
|
payment_method: PaymentMethod
|
|
notes: Optional[str] = None
|
|
|
|
|
|
class PaymentCreate(PaymentBase):
|
|
membership_id: Optional[int] = None
|
|
|
|
|
|
class PaymentUpdate(UTCBaseModel):
|
|
status: Optional[PaymentStatus] = None
|
|
transaction_id: Optional[str] = None
|
|
payment_date: Optional[datetime] = None
|
|
notes: Optional[str] = None
|
|
|
|
|
|
class PaymentResponse(UTCBaseModel):
|
|
model_config = ConfigDict(from_attributes=True)
|
|
|
|
id: int
|
|
user_id: int
|
|
membership_id: Optional[int] = None
|
|
amount: float
|
|
payment_method: PaymentMethod
|
|
status: PaymentStatus
|
|
transaction_id: Optional[str] = None
|
|
payment_date: Optional[datetime] = None
|
|
notes: Optional[str] = None
|
|
created_at: datetime
|
|
|
|
|
|
# Square Payment Schemas
|
|
class SquarePaymentRequest(UTCBaseModel):
|
|
"""Request schema for Square payment processing"""
|
|
source_id: str = Field(..., description="Payment source ID from Square Web Payments SDK")
|
|
tier_id: int = Field(..., description="Membership tier ID to create membership for")
|
|
amount: float = Field(..., gt=0, description="Payment amount in GBP")
|
|
idempotency_key: Optional[str] = Field(None, description="Unique key to prevent duplicate payments")
|
|
note: Optional[str] = Field(None, description="Optional payment note")
|
|
billing_details: Optional[dict] = Field(None, description="Billing address and cardholder name for AVS")
|
|
|
|
|
|
class SquarePaymentResponse(UTCBaseModel):
|
|
"""Response schema for Square payment"""
|
|
success: bool
|
|
payment_id: Optional[str] = None
|
|
status: Optional[str] = None
|
|
amount: Optional[float] = None
|
|
currency: Optional[str] = None
|
|
receipt_url: Optional[str] = None
|
|
errors: Optional[list[str]] = None
|
|
database_payment_id: Optional[int] = None
|
|
membership_id: Optional[int] = Field(None, description="Created membership ID")
|
|
|
|
|
|
class SquareRefundRequest(UTCBaseModel):
|
|
"""Request schema for Square payment refund"""
|
|
payment_id: int = Field(..., description="Database payment ID")
|
|
amount: Optional[float] = Field(None, gt=0, description="Amount to refund (None for full refund)")
|
|
reason: Optional[str] = Field(None, description="Reason for refund")
|
|
|
|
|
|
# Message Response
|
|
class MessageResponse(UTCBaseModel):
|
|
message: str
|
|
detail: Optional[str] = None
|
|
|
|
|
|
# Email Template Schemas
|
|
class EmailTemplateBase(UTCBaseModel):
|
|
template_key: str
|
|
name: str
|
|
subject: str
|
|
html_body: str
|
|
text_body: Optional[str] = None
|
|
variables: Optional[str] = None
|
|
|
|
|
|
class EmailTemplateCreate(EmailTemplateBase):
|
|
pass
|
|
|
|
|
|
class EmailTemplateUpdate(UTCBaseModel):
|
|
name: Optional[str] = None
|
|
subject: Optional[str] = None
|
|
html_body: Optional[str] = None
|
|
text_body: Optional[str] = None
|
|
variables: Optional[str] = None
|
|
is_active: Optional[bool] = None
|
|
|
|
|
|
class EmailTemplateResponse(EmailTemplateBase):
|
|
model_config = ConfigDict(from_attributes=True)
|
|
|
|
id: int
|
|
is_active: bool
|
|
created_at: datetime
|
|
updated_at: datetime
|
|
|
|
|
|
# Event Schemas
|
|
class EventBase(UTCBaseModel):
|
|
title: str = Field(..., min_length=1, max_length=255)
|
|
description: Optional[str] = None
|
|
event_date: datetime
|
|
event_time: Optional[str] = Field(None, pattern=r'^([01]?[0-9]|2[0-3]):[0-5][0-9]$')
|
|
location: Optional[str] = None
|
|
max_attendees: Optional[int] = Field(None, gt=0)
|
|
|
|
|
|
class EventCreate(EventBase):
|
|
pass
|
|
|
|
|
|
class EventUpdate(UTCBaseModel):
|
|
title: Optional[str] = Field(None, min_length=1, max_length=255)
|
|
description: Optional[str] = None
|
|
event_date: Optional[datetime] = None
|
|
event_time: Optional[str] = Field(None, pattern=r'^([01]?[0-9]|2[0-3]):[0-5][0-9]$')
|
|
location: Optional[str] = None
|
|
max_attendees: Optional[int] = Field(None, gt=0)
|
|
status: Optional[str] = None
|
|
|
|
|
|
class EventResponse(EventBase):
|
|
model_config = ConfigDict(from_attributes=True)
|
|
|
|
id: int
|
|
status: str
|
|
created_by: int
|
|
created_at: datetime
|
|
updated_at: datetime
|
|
|
|
|
|
# Event RSVP Schemas
|
|
class EventRSVPBase(UTCBaseModel):
|
|
status: str = Field(..., pattern="^(pending|attending|not_attending|maybe)$")
|
|
notes: Optional[str] = None
|
|
|
|
|
|
class EventRSVPUpdate(EventRSVPBase):
|
|
pass
|
|
|
|
|
|
class EventRSVPResponse(EventRSVPBase):
|
|
model_config = ConfigDict(from_attributes=True)
|
|
|
|
id: int
|
|
event_id: int
|
|
user_id: int
|
|
attended: bool
|
|
created_at: datetime
|
|
updated_at: datetime
|
|
|
|
|
|
# Profile Question Schemas
|
|
ProfileQuestionInputType = Literal["text", "number", "boolean", "date", "select"]
|
|
|
|
|
|
class QuestionOption(UTCBaseModel):
|
|
label: str = Field(..., min_length=1, max_length=100)
|
|
value: str = Field(..., min_length=1, max_length=100)
|
|
|
|
|
|
class ProfileQuestionBase(UTCBaseModel):
|
|
key: str = Field(..., min_length=2, max_length=100, pattern=r"^[a-z0-9_]+$")
|
|
label: str = Field(..., min_length=2, max_length=255)
|
|
help_text: Optional[str] = None
|
|
input_type: ProfileQuestionInputType
|
|
placeholder: Optional[str] = Field(None, max_length=255)
|
|
options: Optional[list[QuestionOption]] = None
|
|
is_required: bool = False
|
|
is_active: bool = True
|
|
admin_only_edit: bool = False
|
|
display_order: int = 0
|
|
depends_on_question_id: Optional[int] = None
|
|
depends_on_value: Optional[str] = Field(None, max_length=255)
|
|
|
|
|
|
class ProfileQuestionCreate(ProfileQuestionBase):
|
|
pass
|
|
|
|
|
|
class ProfileQuestionUpdate(UTCBaseModel):
|
|
key: Optional[str] = Field(None, min_length=2, max_length=100, pattern=r"^[a-z0-9_]+$")
|
|
label: Optional[str] = Field(None, min_length=2, max_length=255)
|
|
help_text: Optional[str] = None
|
|
input_type: Optional[ProfileQuestionInputType] = None
|
|
placeholder: Optional[str] = Field(None, max_length=255)
|
|
options: Optional[list[QuestionOption]] = None
|
|
is_required: Optional[bool] = None
|
|
is_active: Optional[bool] = None
|
|
admin_only_edit: Optional[bool] = None
|
|
display_order: Optional[int] = None
|
|
depends_on_question_id: Optional[int] = None
|
|
depends_on_value: Optional[str] = Field(None, max_length=255)
|
|
|
|
|
|
class ProfileQuestionResponse(UTCBaseModel):
|
|
model_config = ConfigDict(from_attributes=True)
|
|
|
|
id: int
|
|
key: str
|
|
label: str
|
|
help_text: Optional[str] = None
|
|
input_type: ProfileQuestionInputType
|
|
placeholder: Optional[str] = None
|
|
options: list[QuestionOption] = []
|
|
is_required: bool
|
|
is_active: bool
|
|
admin_only_edit: bool
|
|
display_order: int
|
|
depends_on_question_id: Optional[int] = None
|
|
depends_on_value: Optional[str] = None
|
|
created_at: datetime
|
|
updated_at: datetime
|
|
|
|
|
|
class ProfileQuestionForUser(ProfileQuestionResponse):
|
|
answer: Optional[Any] = None
|
|
can_edit: bool = True
|
|
|
|
|
|
class ProfileAnswerUpdate(UTCBaseModel):
|
|
question_id: int
|
|
value: Optional[Any] = None
|
|
|
|
|
|
class ProfileAnswersUpdateRequest(UTCBaseModel):
|
|
answers: list[ProfileAnswerUpdate]
|
|
|
|
|
|
# ESP RFID Reader Schemas
|
|
class EspReaderBase(UTCBaseModel):
|
|
device_id: str = Field(..., min_length=2, max_length=100, pattern=r"^[A-Za-z0-9_.:-]+$")
|
|
name: str = Field(..., min_length=1, max_length=255)
|
|
location: Optional[str] = Field(None, max_length=255)
|
|
reader_type: EspReaderType = EspReaderType.CHECKIN_CHECKOUT
|
|
notes: Optional[str] = None
|
|
is_active: bool = True
|
|
can_write_cards: bool = False
|
|
firmware_version: Optional[str] = Field(None, max_length=100)
|
|
|
|
|
|
class EspReaderCreate(EspReaderBase):
|
|
api_key: Optional[str] = Field(None, min_length=16, max_length=255)
|
|
|
|
|
|
class EspReaderUpdate(UTCBaseModel):
|
|
name: Optional[str] = Field(None, min_length=1, max_length=255)
|
|
location: Optional[str] = Field(None, max_length=255)
|
|
reader_type: Optional[EspReaderType] = None
|
|
notes: Optional[str] = None
|
|
is_active: Optional[bool] = None
|
|
can_write_cards: Optional[bool] = None
|
|
rotate_api_key: bool = False
|
|
|
|
|
|
class EspReaderResponse(EspReaderBase):
|
|
model_config = ConfigDict(from_attributes=True)
|
|
|
|
id: int
|
|
provisioning_status: EspReaderProvisioningStatus
|
|
last_seen_at: Optional[datetime] = None
|
|
approved_at: Optional[datetime] = None
|
|
provisioned_at: Optional[datetime] = None
|
|
created_at: datetime
|
|
updated_at: datetime
|
|
|
|
|
|
class EspReaderCreateResponse(EspReaderResponse):
|
|
api_key: str
|
|
|
|
|
|
class EspReaderRegistrationRequest(UTCBaseModel):
|
|
device_id: str = Field(..., min_length=2, max_length=100, pattern=r"^[A-Za-z0-9_.:-]+$")
|
|
name: str = Field(..., min_length=1, max_length=255)
|
|
location: Optional[str] = Field(None, max_length=255)
|
|
reader_type: EspReaderType = EspReaderType.CHECKIN_CHECKOUT
|
|
can_write_cards: bool = False
|
|
firmware_version: Optional[str] = Field(None, max_length=100)
|
|
notes: Optional[str] = None
|
|
|
|
|
|
class EspReaderRegistrationResponse(UTCBaseModel):
|
|
device_id: str
|
|
provisioning_status: EspReaderProvisioningStatus
|
|
registration_token: str
|
|
message: str
|
|
poll_interval_seconds: int = 5
|
|
|
|
|
|
class EspReaderProvisioningResponse(UTCBaseModel):
|
|
device_id: str
|
|
provisioning_status: EspReaderProvisioningStatus
|
|
message: str
|
|
api_key: Optional[str] = None
|
|
apiKey: Optional[str] = None
|
|
poll_interval_seconds: int = 5
|
|
|
|
|
|
class RfidCardBase(UTCBaseModel):
|
|
uid: str = Field(..., min_length=2, max_length=100)
|
|
user_id: Optional[int] = None
|
|
label: Optional[str] = Field(None, max_length=255)
|
|
is_active: bool = True
|
|
|
|
|
|
class RfidCardCreate(RfidCardBase):
|
|
pass
|
|
|
|
|
|
class RfidCardUpdate(UTCBaseModel):
|
|
user_id: Optional[int] = None
|
|
label: Optional[str] = Field(None, max_length=255)
|
|
is_active: Optional[bool] = None
|
|
|
|
|
|
class RfidCardResponse(RfidCardBase):
|
|
model_config = ConfigDict(from_attributes=True)
|
|
|
|
id: int
|
|
created_at: datetime
|
|
updated_at: datetime
|
|
|
|
|
|
class RfidTapRequest(UTCBaseModel):
|
|
card_uid: str = Field(..., min_length=2, max_length=100)
|
|
tapped_at: Optional[datetime] = None
|
|
reader_type: Optional[EspReaderType] = None
|
|
|
|
|
|
class RfidTapResponse(UTCBaseModel):
|
|
accepted: bool
|
|
action: EspTapAction
|
|
message: str
|
|
server_time_utc: datetime
|
|
tap_id: int
|
|
session_id: Optional[int] = None
|
|
user_id: Optional[int] = None
|
|
user_name: Optional[str] = None
|
|
checked_in_at: Optional[datetime] = None
|
|
checked_out_at: Optional[datetime] = None
|
|
duration_seconds: Optional[int] = None
|
|
|
|
|
|
class RfidWriteJobCreate(UTCBaseModel):
|
|
reader_id: int
|
|
user_id: int
|
|
label: str = Field(..., min_length=1, max_length=255)
|
|
|
|
|
|
class RfidWriteJobCompleteRequest(UTCBaseModel):
|
|
card_uid: Optional[str] = Field(None, min_length=2, max_length=100)
|
|
success: bool
|
|
error_message: Optional[str] = Field(None, max_length=500)
|
|
|
|
|
|
class RfidWriteJobResponse(UTCBaseModel):
|
|
model_config = ConfigDict(from_attributes=True)
|
|
|
|
id: int
|
|
reader_id: int
|
|
user_id: int
|
|
card_id: Optional[int] = None
|
|
label: str
|
|
status: RfidWriteJobStatus
|
|
requested_by_user_id: int
|
|
card_uid: Optional[str] = None
|
|
write_payload: Optional[str] = None
|
|
claimed_at: Optional[datetime] = None
|
|
completed_at: Optional[datetime] = None
|
|
error_message: Optional[str] = None
|
|
created_at: datetime
|
|
updated_at: datetime
|
|
|
|
|
|
class EspTimeResponse(UTCBaseModel):
|
|
server_time_utc: datetime
|
|
unix_ms: int
|
|
poll_interval_seconds: int = 3
|
|
|
|
|
|
class EspHeartbeatRequest(UTCBaseModel):
|
|
mode: str = Field(..., max_length=50)
|
|
message: Optional[str] = Field(None, max_length=255)
|
|
wifi_rssi: Optional[int] = None
|
|
free_heap: Optional[int] = None
|
|
firmware_version: Optional[str] = Field(None, max_length=100)
|
|
active_write_job_id: Optional[int] = None
|
|
|
|
|
|
class EspHeartbeatResponse(UTCBaseModel):
|
|
ok: bool
|
|
server_time_utc: datetime
|
|
unix_ms: int
|
|
heartbeat_interval_seconds: int = 10
|
|
time_poll_interval_seconds: int = 3
|
|
write_job_poll_interval_seconds: int = 3
|
|
|
|
|
|
class EspDashboardLoginResponse(UTCBaseModel):
|
|
valid: bool
|
|
user_id: Optional[int] = None
|
|
role: Optional[UserRole] = None
|
|
user_name: Optional[str] = None
|
|
|
|
|
|
class RfidTapAdminResponse(UTCBaseModel):
|
|
model_config = ConfigDict(from_attributes=True)
|
|
|
|
id: int
|
|
reader_id: int
|
|
card_id: Optional[int] = None
|
|
user_id: Optional[int] = None
|
|
card_uid: str
|
|
action: EspTapAction
|
|
accepted: bool
|
|
message: Optional[str] = None
|
|
tapped_at: datetime
|
|
created_at: datetime
|
|
|
|
|
|
class AttendanceSessionResponse(UTCBaseModel):
|
|
model_config = ConfigDict(from_attributes=True)
|
|
|
|
id: int
|
|
user_id: int
|
|
reader_id: int
|
|
check_in_tap_id: int
|
|
check_out_tap_id: Optional[int] = None
|
|
checked_in_at: datetime
|
|
checked_out_at: Optional[datetime] = None
|
|
checkout_source: Optional[str] = None
|
|
system_flag_reason: Optional[str] = None
|
|
duration_seconds: Optional[int] = None
|
|
is_open: bool
|
|
created_at: datetime
|
|
updated_at: datetime
|
|
|
|
|
|
class StaleSessionCloseRequest(UTCBaseModel):
|
|
cutoff_date: Optional[date] = None
|
|
checkout_hour: int = Field(17, ge=0, le=23)
|
|
|
|
|
|
class StaleSessionCloseResponse(UTCBaseModel):
|
|
closed_count: int
|