1468af1391
- Created FastAPI application with SQLite database integration - Implemented OpenWeatherMap client with caching - Added endpoints for current weather, forecasts, and history - Included comprehensive error handling and validation - Set up Alembic migrations - Created detailed README with usage examples generated with BackendIM... (backend.im)
64 lines
2.1 KiB
Python
64 lines
2.1 KiB
Python
from typing import Any, Dict, Generic, List, Optional, Type, TypeVar, Union
|
|
from fastapi.encoders import jsonable_encoder
|
|
from pydantic import BaseModel
|
|
from sqlalchemy.orm import Session
|
|
from sqlalchemy import desc
|
|
|
|
from app.database.session import Base
|
|
|
|
ModelType = TypeVar("ModelType", bound=Base)
|
|
CreateSchemaType = TypeVar("CreateSchemaType", bound=BaseModel)
|
|
UpdateSchemaType = TypeVar("UpdateSchemaType", bound=BaseModel)
|
|
|
|
class CRUDBase(Generic[ModelType, CreateSchemaType, UpdateSchemaType]):
|
|
def __init__(self, model: Type[ModelType]):
|
|
"""
|
|
CRUD object with default methods to Create, Read, Update, Delete (CRUD).
|
|
|
|
**Parameters**
|
|
* `model`: A SQLAlchemy model class
|
|
* `schema`: A Pydantic model (schema) class
|
|
"""
|
|
self.model = model
|
|
|
|
def get(self, db: Session, id: Any) -> Optional[ModelType]:
|
|
return db.query(self.model).filter(self.model.id == id).first()
|
|
|
|
def get_multi(
|
|
self, db: Session, *, skip: int = 0, limit: int = 100
|
|
) -> List[ModelType]:
|
|
return db.query(self.model).offset(skip).limit(limit).all()
|
|
|
|
def create(self, db: Session, *, obj_in: CreateSchemaType) -> ModelType:
|
|
obj_in_data = jsonable_encoder(obj_in)
|
|
db_obj = self.model(**obj_in_data)
|
|
db.add(db_obj)
|
|
db.commit()
|
|
db.refresh(db_obj)
|
|
return db_obj
|
|
|
|
def update(
|
|
self,
|
|
db: Session,
|
|
*,
|
|
db_obj: ModelType,
|
|
obj_in: Union[UpdateSchemaType, Dict[str, Any]]
|
|
) -> ModelType:
|
|
obj_data = jsonable_encoder(db_obj)
|
|
if isinstance(obj_in, dict):
|
|
update_data = obj_in
|
|
else:
|
|
update_data = obj_in.dict(exclude_unset=True)
|
|
for field in obj_data:
|
|
if field in update_data:
|
|
setattr(db_obj, field, update_data[field])
|
|
db.add(db_obj)
|
|
db.commit()
|
|
db.refresh(db_obj)
|
|
return db_obj
|
|
|
|
def remove(self, db: Session, *, id: int) -> ModelType:
|
|
obj = db.query(self.model).get(id)
|
|
db.delete(obj)
|
|
db.commit()
|
|
return obj |