741f301b11
- Set up project structure with FastAPI - Implement user authentication system with JWT tokens - Create database models for users, notes, and collections - Set up SQLAlchemy ORM and Alembic migrations - Implement CRUD operations for notes and collections - Add filtering and sorting capabilities for notes - Implement health check endpoint - Update project documentation
60 lines
1.5 KiB
Python
60 lines
1.5 KiB
Python
from typing import Any
|
|
|
|
from sqlalchemy.orm import Session
|
|
|
|
from app.models.collection import Collection
|
|
from app.schemas.collection import CollectionCreate, CollectionUpdate
|
|
|
|
|
|
def get(db: Session, collection_id: int) -> Collection | None:
|
|
return db.query(Collection).filter(Collection.id == collection_id).first()
|
|
|
|
|
|
def get_multi_by_owner(
|
|
db: Session, owner_id: int, skip: int = 0, limit: int = 100
|
|
) -> list[Collection]:
|
|
return (
|
|
db.query(Collection)
|
|
.filter(Collection.owner_id == owner_id)
|
|
.offset(skip)
|
|
.limit(limit)
|
|
.all()
|
|
)
|
|
|
|
|
|
def create(db: Session, *, obj_in: CollectionCreate, owner_id: int) -> Collection:
|
|
db_obj = Collection(
|
|
**obj_in.model_dump(),
|
|
owner_id=owner_id,
|
|
)
|
|
db.add(db_obj)
|
|
db.commit()
|
|
db.refresh(db_obj)
|
|
return db_obj
|
|
|
|
|
|
def update(
|
|
db: Session, *, db_obj: Collection, obj_in: CollectionUpdate | dict[str, Any]
|
|
) -> Collection:
|
|
if isinstance(obj_in, dict):
|
|
update_data = obj_in
|
|
else:
|
|
update_data = obj_in.model_dump(exclude_unset=True)
|
|
for field in update_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(db: Session, *, collection_id: int, owner_id: int) -> Collection | None:
|
|
obj = db.query(Collection).filter(
|
|
Collection.id == collection_id, Collection.owner_id == owner_id
|
|
).first()
|
|
if obj:
|
|
db.delete(obj)
|
|
db.commit()
|
|
return obj
|