db/herodb/src/models/py/api.py

#!/usr/bin/env python3
"""
FastAPI server providing CRUD operations for business models.
"""
import os
from datetime import datetime
from typing import List, Optional

from fastapi import Depends, FastAPI, HTTPException, Query
from fastapi.middleware.cors import CORSMiddleware
from sqlmodel import Session, SQLModel, create_engine, select

from models import (
    Currency,
    Customer,
    Product,
    ProductComponent,
    ProductStatus,
    ProductType,
    Sale,
    SaleItem,
    SaleStatus,
)

# Create database
DATABASE_URL = os.environ.get("DATABASE_URL", "sqlite:///business.db")
engine = create_engine(DATABASE_URL, echo=False)

# Create tables
SQLModel.metadata.create_all(engine)

# Create FastAPI app
app = FastAPI(
    title="Business API",
    description="API for business models",
    version="1.0.0",
    docs_url="/docs",
    redoc_url="/redoc",
    openapi_url="/openapi.json",
)

# Add CORS middleware
app.add_middleware(
    CORSMiddleware,
    allow_origins=["*"],
    allow_credentials=True,
    allow_methods=["*"],
    allow_headers=["*"],
)

# Dependency to get database session
def get_session():
    with Session(engine) as session:
        yield session


# Root endpoint
@app.get("/")
async def root():
    return {"message": "Welcome to the Business API"}


# Currency endpoints
@app.post("/currencies/", response_model=Currency, tags=["Currencies"])
def create_currency(currency: Currency, session: Session = Depends(get_session)):
    """Create a new currency"""
    session.add(currency)
    session.commit()
    session.refresh(currency)
    return currency


@app.get("/currencies/", response_model=List[Currency], tags=["Currencies"])
def read_currencies(
    skip: int = 0, limit: int = 100, session: Session = Depends(get_session)
):
    """Get all currencies"""
    currencies = session.exec(select(Currency).offset(skip).limit(limit)).all()
    return currencies


@app.get("/currencies/{currency_id}", response_model=Currency, tags=["Currencies"])
def read_currency(currency_id: int, session: Session = Depends(get_session)):
    """Get a currency by ID"""
    currency = session.get(Currency, currency_id)
    if not currency:
        raise HTTPException(status_code=404, detail="Currency not found")
    return currency


@app.put("/currencies/{currency_id}", response_model=Currency, tags=["Currencies"])
def update_currency(
    currency_id: int, currency_data: Currency, session: Session = Depends(get_session)
):
    """Update a currency"""
    currency = session.get(Currency, currency_id)
    if not currency:
        raise HTTPException(status_code=404, detail="Currency not found")
    
    # Update currency attributes
    currency_data_dict = currency_data.dict(exclude_unset=True)
    for key, value in currency_data_dict.items():
        setattr(currency, key, value)
    
    session.add(currency)
    session.commit()
    session.refresh(currency)
    return currency


@app.delete("/currencies/{currency_id}", tags=["Currencies"])
def delete_currency(currency_id: int, session: Session = Depends(get_session)):
    """Delete a currency"""
    currency = session.get(Currency, currency_id)
    if not currency:
        raise HTTPException(status_code=404, detail="Currency not found")
    
    session.delete(currency)
    session.commit()
    return {"message": "Currency deleted successfully"}


# Customer endpoints
@app.post("/customers/", response_model=Customer, tags=["Customers"])
def create_customer(customer: Customer, session: Session = Depends(get_session)):
    """Create a new customer"""
    session.add(customer)
    session.commit()
    session.refresh(customer)
    return customer


@app.get("/customers/", response_model=List[Customer], tags=["Customers"])
def read_customers(
    skip: int = 0, limit: int = 100, session: Session = Depends(get_session)
):
    """Get all customers"""
    customers = session.exec(select(Customer).offset(skip).limit(limit)).all()
    return customers


@app.get("/customers/{customer_id}", response_model=Customer, tags=["Customers"])
def read_customer(customer_id: int, session: Session = Depends(get_session)):
    """Get a customer by ID"""
    customer = session.get(Customer, customer_id)
    if not customer:
        raise HTTPException(status_code=404, detail="Customer not found")
    return customer


@app.put("/customers/{customer_id}", response_model=Customer, tags=["Customers"])
def update_customer(
    customer_id: int, customer_data: Customer, session: Session = Depends(get_session)
):
    """Update a customer"""
    customer = session.get(Customer, customer_id)
    if not customer:
        raise HTTPException(status_code=404, detail="Customer not found")
    
    # Update customer attributes
    customer_data_dict = customer_data.dict(exclude_unset=True)
    for key, value in customer_data_dict.items():
        setattr(customer, key, value)
    
    session.add(customer)
    session.commit()
    session.refresh(customer)
    return customer


@app.delete("/customers/{customer_id}", tags=["Customers"])
def delete_customer(customer_id: int, session: Session = Depends(get_session)):
    """Delete a customer"""
    customer = session.get(Customer, customer_id)
    if not customer:
        raise HTTPException(status_code=404, detail="Customer not found")
    
    session.delete(customer)
    session.commit()
    return {"message": "Customer deleted successfully"}


# Product endpoints
@app.post("/products/", response_model=Product, tags=["Products"])
def create_product(product: Product, session: Session = Depends(get_session)):
    """Create a new product"""
    session.add(product)
    session.commit()
    session.refresh(product)
    return product


@app.get("/products/", response_model=List[Product], tags=["Products"])
def read_products(
    skip: int = 0,
    limit: int = 100,
    category: Optional[str] = None,
    status: Optional[ProductStatus] = None,
    istemplate: Optional[bool] = None,
    session: Session = Depends(get_session)
):
    """Get all products with optional filtering"""
    query = select(Product)
    
    if category:
        query = query.where(Product.category == category)
    
    if status:
        query = query.where(Product.status == status)
    
    if istemplate is not None:
        query = query.where(Product.istemplate == istemplate)
    
    products = session.exec(query.offset(skip).limit(limit)).all()
    return products


@app.get("/products/{product_id}", response_model=Product, tags=["Products"])
def read_product(product_id: int, session: Session = Depends(get_session)):
    """Get a product by ID"""
    product = session.get(Product, product_id)
    if not product:
        raise HTTPException(status_code=404, detail="Product not found")
    return product


@app.put("/products/{product_id}", response_model=Product, tags=["Products"])
def update_product(
    product_id: int, product_data: Product, session: Session = Depends(get_session)
):
    """Update a product"""
    product = session.get(Product, product_id)
    if not product:
        raise HTTPException(status_code=404, detail="Product not found")
    
    # Update product attributes
    product_data_dict = product_data.dict(exclude_unset=True)
    for key, value in product_data_dict.items():
        setattr(product, key, value)
    
    session.add(product)
    session.commit()
    session.refresh(product)
    return product


@app.delete("/products/{product_id}", tags=["Products"])
def delete_product(product_id: int, session: Session = Depends(get_session)):
    """Delete a product"""
    product = session.get(Product, product_id)
    if not product:
        raise HTTPException(status_code=404, detail="Product not found")
    
    session.delete(product)
    session.commit()
    return {"message": "Product deleted successfully"}


# Product Component endpoints
@app.post("/products/{product_id}/components/", response_model=ProductComponent, tags=["Product Components"])
def create_product_component(
    product_id: int, component: ProductComponent, session: Session = Depends(get_session)
):
    """Add a component to a product"""
    product = session.get(Product, product_id)
    if not product:
        raise HTTPException(status_code=404, detail="Product not found")
    
    component.product_id = product_id
    session.add(component)
    session.commit()
    session.refresh(component)
    return component


@app.get("/products/{product_id}/components/", response_model=List[ProductComponent], tags=["Product Components"])
def read_product_components(
    product_id: int, session: Session = Depends(get_session)
):
    """Get all components for a product"""
    product = session.get(Product, product_id)
    if not product:
        raise HTTPException(status_code=404, detail="Product not found")
    
    return product.components


# Sale endpoints
@app.post("/sales/", response_model=Sale, tags=["Sales"])
def create_sale(sale: Sale, session: Session = Depends(get_session)):
    """Create a new sale"""
    # Ensure customer exists if customer_id is provided
    if sale.customer_id:
        customer = session.get(Customer, sale.customer_id)
        if not customer:
            raise HTTPException(status_code=404, detail="Customer not found")
    
    session.add(sale)
    session.commit()
    session.refresh(sale)
    return sale


@app.get("/sales/", response_model=List[Sale], tags=["Sales"])
def read_sales(
    skip: int = 0, 
    limit: int = 100, 
    status: Optional[SaleStatus] = None,
    customer_id: Optional[int] = None,
    session: Session = Depends(get_session)
):
    """Get all sales with optional filtering"""
    query = select(Sale)
    
    if status:
        query = query.where(Sale.status == status)
    
    if customer_id:
        query = query.where(Sale.customer_id == customer_id)
    
    sales = session.exec(query.offset(skip).limit(limit)).all()
    return sales


@app.get("/sales/{sale_id}", response_model=Sale, tags=["Sales"])
def read_sale(sale_id: int, session: Session = Depends(get_session)):
    """Get a sale by ID"""
    sale = session.get(Sale, sale_id)
    if not sale:
        raise HTTPException(status_code=404, detail="Sale not found")
    return sale


@app.put("/sales/{sale_id}", response_model=Sale, tags=["Sales"])
def update_sale(
    sale_id: int, sale_data: Sale, session: Session = Depends(get_session)
):
    """Update a sale"""
    sale = session.get(Sale, sale_id)
    if not sale:
        raise HTTPException(status_code=404, detail="Sale not found")
    
    # Update sale attributes
    sale_data_dict = sale_data.dict(exclude_unset=True)
    for key, value in sale_data_dict.items():
        setattr(sale, key, value)
    
    session.add(sale)
    session.commit()
    session.refresh(sale)
    return sale


@app.delete("/sales/{sale_id}", tags=["Sales"])
def delete_sale(sale_id: int, session: Session = Depends(get_session)):
    """Delete a sale"""
    sale = session.get(Sale, sale_id)
    if not sale:
        raise HTTPException(status_code=404, detail="Sale not found")
    
    session.delete(sale)
    session.commit()
    return {"message": "Sale deleted successfully"}


# Sale Item endpoints
@app.post("/sales/{sale_id}/items/", response_model=SaleItem, tags=["Sale Items"])
def create_sale_item(
    sale_id: int, item: SaleItem, session: Session = Depends(get_session)
):
    """Add an item to a sale"""
    sale = session.get(Sale, sale_id)
    if not sale:
        raise HTTPException(status_code=404, detail="Sale not found")
    
    item.sale_id = sale_id
    session.add(item)
    session.commit()
    session.refresh(item)
    
    # Update the sale's total amount
    sale.add_item(item)
    session.add(sale)
    session.commit()
    
    return item


@app.get("/sales/{sale_id}/items/", response_model=List[SaleItem], tags=["Sale Items"])
def read_sale_items(
    sale_id: int, session: Session = Depends(get_session)
):
    """Get all items for a sale"""
    sale = session.get(Sale, sale_id)
    if not sale:
        raise HTTPException(status_code=404, detail="Sale not found")
    
    return sale.items


# Convenience endpoints
@app.put("/sales/{sale_id}/status/{status}", response_model=Sale, tags=["Convenience"])
def update_sale_status(
    sale_id: int, status: SaleStatus, session: Session = Depends(get_session)
):
    """Update the status of a sale"""
    sale = session.get(Sale, sale_id)
    if not sale:
        raise HTTPException(status_code=404, detail="Sale not found")
    
    sale.update_status(status)
    session.add(sale)
    session.commit()
    session.refresh(sale)
    return sale


@app.get("/products/available/", response_model=List[Product], tags=["Convenience"])
def get_available_products(
    istemplate: Optional[bool] = False,
    session: Session = Depends(get_session)
):
    """Get all available products"""
    query = select(Product).where(
        Product.status == ProductStatus.AVAILABLE,
        Product.purchase_till > datetime.utcnow(),
        Product.istemplate == istemplate
    )
    products = session.exec(query).all()
    return products


@app.get("/customers/{customer_id}/sales/", response_model=List[Sale], tags=["Convenience"])
def get_customer_sales(
    customer_id: int, 
    status: Optional[SaleStatus] = None,
    session: Session = Depends(get_session)
):
    """Get all sales for a customer"""
    customer = session.get(Customer, customer_id)
    if not customer:
        raise HTTPException(status_code=404, detail="Customer not found")
    
    query = select(Sale).where(Sale.customer_id == customer_id)
    
    if status:
        query = query.where(Sale.status == status)
    
    sales = session.exec(query).all()
    return sales


if __name__ == "__main__":
    import uvicorn
    uvicorn.run(app, host="0.0.0.0", port=8000)