|
|
|
from pydantic import BaseModel, UUID4, Field, ConfigDict, condecimal, constr, validator |
|
from datetime import datetime |
|
from typing import Optional, Dict |
|
import uuid |
|
|
|
from pydantic import BaseModel, UUID4, Field, validator, ConfigDict |
|
from uuid import uuid4, UUID |
|
|
|
|
|
class TransactionSchema(BaseModel): |
|
transaction_id: str |
|
amount_received: float |
|
phone_number: str |
|
name: str |
|
date: datetime |
|
new_balance: float |
|
|
|
|
|
class PayloadSchema(BaseModel): |
|
message: str |
|
phoneNumber: str |
|
receivedAt: datetime |
|
simNumber: Optional[int] |
|
|
|
|
|
class MessageCreate(BaseModel): |
|
deviceId: Optional[str] |
|
event: Optional[str] |
|
id: Optional[str] |
|
payload: PayloadSchema |
|
webhookId: Optional[str] |
|
|
|
|
|
class MessageResponse(BaseModel): |
|
id: Optional[UUID4] = Field( |
|
default_factory=uuid.uuid4 |
|
) |
|
device_id: Optional[str] = None |
|
event: Optional[str] = None |
|
message_id: Optional[str] = None |
|
webhook_id: Optional[str] = None |
|
message_content: Optional[str] = None |
|
phone_number: Optional[str] = None |
|
received_at: Optional[datetime] = None |
|
sim_number: Optional[int] = None |
|
parsed_data: Optional[Dict] = None |
|
created_time: Optional[datetime] = None |
|
model_config = ConfigDict(from_attributes=True) |
|
|
|
@validator("id", pre=True, always=True) |
|
def validate_uuid(cls, v): |
|
""" |
|
Ensure all ID fields are valid UUIDs. If the input is invalid or missing, |
|
generate a new UUID. |
|
""" |
|
if v is None or not isinstance(v, (UUID, str)): |
|
return uuid4() |
|
try: |
|
return UUID(v) |
|
except ValueError: |
|
return uuid4() |
|
|
|
|
|
|
|
|
|
|