125 lines
3.4 KiB
Python
125 lines
3.4 KiB
Python
import logging
|
|
from fastapi import FastAPI, HTTPException
|
|
from pymongo import MongoClient
|
|
from bson import ObjectId
|
|
from pydantic import BaseModel, Field
|
|
from typing import List, Optional
|
|
|
|
# Configure logging
|
|
logging.basicConfig(level=logging.DEBUG,
|
|
format='%(asctime)s - %(levelname)s - %(message)s')
|
|
logger = logging.getLogger(__name__)
|
|
|
|
app = FastAPI()
|
|
logger.debug("FastAPI application initialized")
|
|
|
|
# MongoDB connection
|
|
client = MongoClient("mongodb://localhost:27017")
|
|
db = client["pos_system"]
|
|
logger.debug("MongoDB connection established")
|
|
|
|
# Models
|
|
|
|
|
|
class Item(BaseModel):
|
|
name: str
|
|
price: float
|
|
quantity: int
|
|
unit: str
|
|
related_items: List[str] = []
|
|
|
|
|
|
class Deal(BaseModel):
|
|
name: str
|
|
items: List[str]
|
|
discount: float
|
|
|
|
|
|
class Order(BaseModel):
|
|
customer_name: str
|
|
items: List[str]
|
|
total_amount: float
|
|
payment_method: str
|
|
date: str
|
|
voucher: Optional[str] = None
|
|
|
|
|
|
logger.debug("Data models defined")
|
|
|
|
# API Routes
|
|
|
|
|
|
@app.post("/items/")
|
|
async def create_item(item: Item):
|
|
logger.debug(f"Received request to create item: {item}")
|
|
result = db.items.insert_one(item.dict())
|
|
logger.debug(f"Item created with ID: {result.inserted_id}")
|
|
return {"id": str(result.inserted_id)}
|
|
|
|
|
|
@app.get("/items/")
|
|
async def read_items():
|
|
logger.debug("Received request to read all items")
|
|
items = list(db.items.find())
|
|
logger.debug(f"Retrieved {len(items)} items")
|
|
return items
|
|
|
|
|
|
@app.get("/items/{item_id}")
|
|
async def read_item(item_id: str):
|
|
logger.debug(f"Received request to read item with ID: {item_id}")
|
|
item = db.items.find_one({"_id": ObjectId(item_id)})
|
|
if item:
|
|
logger.debug(f"Item found: {item}")
|
|
return item
|
|
logger.warning(f"Item with ID {item_id} not found")
|
|
raise HTTPException(status_code=404, detail="Item not found")
|
|
|
|
|
|
@app.put("/items/{item_id}")
|
|
async def update_item(item_id: str, item: Item):
|
|
logger.debug(f"Received request to update item with ID: {item_id}")
|
|
result = db.items.update_one(
|
|
{"_id": ObjectId(item_id)}, {"$set": item.dict()})
|
|
if result.modified_count:
|
|
logger.debug(f"Item with ID {item_id} updated successfully")
|
|
return {"message": "Item updated successfully"}
|
|
logger.warning(f"Item with ID {item_id} not found for update")
|
|
raise HTTPException(status_code=404, detail="Item not found")
|
|
|
|
|
|
@app.delete("/items/{item_id}")
|
|
async def delete_item(item_id: str):
|
|
logger.debug(f"Received request to delete item with ID: {item_id}")
|
|
result = db.items.delete_one({"_id": ObjectId(item_id)})
|
|
if result.deleted_count:
|
|
logger.debug(f"Item with ID {item_id} deleted successfully")
|
|
return {"message": "Item deleted successfully"}
|
|
logger.warning(f"Item with ID {item_id} not found for deletion")
|
|
raise HTTPException(status_code=404, detail="Item not found")
|
|
|
|
# Similar CRUD operations for deals and orders
|
|
|
|
|
|
@app.get("/inventory/")
|
|
async def get_inventory():
|
|
logger.debug("Received request to get inventory")
|
|
inventory = list(db.items.find({}, {"name": 1, "quantity": 1}))
|
|
logger.debug(f"Retrieved inventory with {len(inventory)} items")
|
|
return inventory
|
|
|
|
|
|
@app.get("/sales/")
|
|
async def get_sales_data():
|
|
# Implement aggregation for sales data
|
|
# This could include total sales, popular items, etc.
|
|
pass
|
|
|
|
|
|
if __name__ == "__main__":
|
|
logger.info("Starting the FastAPI application")
|
|
import uvicorn
|
|
uvicorn.run(app, host="0.0.0.0", port=8000)
|
|
|
|
# Add more routes as needed for reporting, analytics, etc.
|