75f5a5a48b
Add FastAPI admin/public API service, database setup, Docker deployment files, docs, and tests.
48 lines
1.6 KiB
Python
48 lines
1.6 KiB
Python
from datetime import datetime
|
|
from sqlalchemy.inspection import inspect
|
|
|
|
|
|
def encode_value(value):
|
|
if isinstance(value, datetime):
|
|
return value.isoformat()
|
|
if isinstance(value, list):
|
|
return [encode_value(item) for item in value]
|
|
if isinstance(value, dict):
|
|
return {key: encode_value(item) for key, item in value.items()}
|
|
return value
|
|
|
|
|
|
def model_dict(instance, include: dict[str, object] | None = None) -> dict:
|
|
data = {column.key: encode_value(getattr(instance, column.key)) for column in inspect(instance).mapper.column_attrs}
|
|
if include:
|
|
for key, value in include.items():
|
|
data[key] = encode_value(value)
|
|
return data
|
|
|
|
|
|
def product_dict(product) -> dict:
|
|
return model_dict(
|
|
product,
|
|
{
|
|
"destination": model_dict(product.destination) if product.destination else None,
|
|
"images": [model_dict(image) for image in sorted(product.images, key=lambda item: item.sortOrder)],
|
|
},
|
|
)
|
|
|
|
|
|
def destination_dict(destination, include_count: bool = False) -> dict:
|
|
data = model_dict(destination, {"aliases": [model_dict(alias) for alias in destination.aliases]})
|
|
if include_count:
|
|
data["_count"] = {"products": len(destination.products)}
|
|
return data
|
|
|
|
|
|
def lead_dict(lead) -> dict:
|
|
return model_dict(
|
|
lead,
|
|
{
|
|
"sourceProduct": {"id": lead.sourceProduct.id, "title": lead.sourceProduct.title} if lead.sourceProduct else None,
|
|
"assignedUser": {"id": lead.assignedUser.id, "name": lead.assignedUser.name} if lead.assignedUser else None,
|
|
},
|
|
)
|