
Implemented: - Basic arithmetic operations (add, subtract, multiply, divide) - Input validation - SQLite database with SQLAlchemy - Alembic migrations - Health endpoint generated with BackendIM... (backend.im)
165 lines
4.3 KiB
Python
165 lines
4.3 KiB
Python
from fastapi import FastAPI, HTTPException, Depends
|
|
from fastapi.responses import JSONResponse
|
|
from pydantic import BaseModel, Field
|
|
from typing import Optional
|
|
from pathlib import Path
|
|
from sqlalchemy import create_engine, Column, Integer, Float, String, DateTime
|
|
from sqlalchemy.ext.declarative import declarative_base
|
|
from sqlalchemy.orm import sessionmaker, Session
|
|
import datetime
|
|
|
|
app = FastAPI(
|
|
title="Calculator API",
|
|
description="A simple calculator API with basic operations",
|
|
version="1.0.0"
|
|
)
|
|
|
|
# Database setup
|
|
DB_DIR = Path("/app") / "storage" / "db"
|
|
DB_DIR.mkdir(parents=True, exist_ok=True)
|
|
|
|
SQLALCHEMY_DATABASE_URL = f"sqlite:///{DB_DIR}/db.sqlite"
|
|
|
|
engine = create_engine(
|
|
SQLALCHEMY_DATABASE_URL,
|
|
connect_args={"check_same_thread": False}
|
|
)
|
|
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
|
|
Base = declarative_base()
|
|
|
|
# Database models
|
|
class Calculation(Base):
|
|
__tablename__ = "calculations"
|
|
|
|
id = Column(Integer, primary_key=True, index=True)
|
|
operation = Column(String, index=True)
|
|
num1 = Column(Float)
|
|
num2 = Column(Float, nullable=True)
|
|
result = Column(Float)
|
|
created_at = Column(DateTime, default=datetime.datetime.utcnow)
|
|
|
|
# Create tables
|
|
Base.metadata.create_all(bind=engine)
|
|
|
|
# Dependency
|
|
def get_db():
|
|
db = SessionLocal()
|
|
try:
|
|
yield db
|
|
finally:
|
|
db.close()
|
|
|
|
# Input models
|
|
class CalculationBase(BaseModel):
|
|
num1: float = Field(..., description="First number")
|
|
num2: float = Field(..., description="Second number")
|
|
|
|
class CalculationResponse(BaseModel):
|
|
result: float = Field(..., description="Result of the calculation")
|
|
operation: str = Field(..., description="Operation performed")
|
|
num1: float = Field(..., description="First number")
|
|
num2: float = Field(..., description="Second number")
|
|
|
|
class Config:
|
|
orm_mode = True
|
|
|
|
# Health check endpoint
|
|
@app.get("/health", tags=["Health"])
|
|
async def health():
|
|
return {"status": "healthy"}
|
|
|
|
# Calculator endpoints
|
|
@app.post("/api/add", response_model=CalculationResponse, tags=["Calculator"])
|
|
async def add(data: CalculationBase, db: Session = Depends(get_db)):
|
|
result = data.num1 + data.num2
|
|
|
|
# Save to database
|
|
db_calc = Calculation(
|
|
operation="add",
|
|
num1=data.num1,
|
|
num2=data.num2,
|
|
result=result
|
|
)
|
|
db.add(db_calc)
|
|
db.commit()
|
|
db.refresh(db_calc)
|
|
|
|
return {
|
|
"result": result,
|
|
"operation": "add",
|
|
"num1": data.num1,
|
|
"num2": data.num2
|
|
}
|
|
|
|
@app.post("/api/subtract", response_model=CalculationResponse, tags=["Calculator"])
|
|
async def subtract(data: CalculationBase, db: Session = Depends(get_db)):
|
|
result = data.num1 - data.num2
|
|
|
|
# Save to database
|
|
db_calc = Calculation(
|
|
operation="subtract",
|
|
num1=data.num1,
|
|
num2=data.num2,
|
|
result=result
|
|
)
|
|
db.add(db_calc)
|
|
db.commit()
|
|
db.refresh(db_calc)
|
|
|
|
return {
|
|
"result": result,
|
|
"operation": "subtract",
|
|
"num1": data.num1,
|
|
"num2": data.num2
|
|
}
|
|
|
|
@app.post("/api/multiply", response_model=CalculationResponse, tags=["Calculator"])
|
|
async def multiply(data: CalculationBase, db: Session = Depends(get_db)):
|
|
result = data.num1 * data.num2
|
|
|
|
# Save to database
|
|
db_calc = Calculation(
|
|
operation="multiply",
|
|
num1=data.num1,
|
|
num2=data.num2,
|
|
result=result
|
|
)
|
|
db.add(db_calc)
|
|
db.commit()
|
|
db.refresh(db_calc)
|
|
|
|
return {
|
|
"result": result,
|
|
"operation": "multiply",
|
|
"num1": data.num1,
|
|
"num2": data.num2
|
|
}
|
|
|
|
@app.post("/api/divide", response_model=CalculationResponse, tags=["Calculator"])
|
|
async def divide(data: CalculationBase, db: Session = Depends(get_db)):
|
|
if data.num2 == 0:
|
|
raise HTTPException(status_code=400, detail="Cannot divide by zero")
|
|
|
|
result = data.num1 / data.num2
|
|
|
|
# Save to database
|
|
db_calc = Calculation(
|
|
operation="divide",
|
|
num1=data.num1,
|
|
num2=data.num2,
|
|
result=result
|
|
)
|
|
db.add(db_calc)
|
|
db.commit()
|
|
db.refresh(db_calc)
|
|
|
|
return {
|
|
"result": result,
|
|
"operation": "divide",
|
|
"num1": data.num1,
|
|
"num2": data.num2
|
|
}
|
|
|
|
if __name__ == "__main__":
|
|
import uvicorn
|
|
uvicorn.run(app, host="0.0.0.0", port=8000) |