curd
"/home/yossef/notes/full-stack/python-backend/fastapi/curd_sql/curd.md"
path: full-stack/python-backend/fastapi/curd_sql/curd.md
- **fileName**: curd
- **Created on**: 2025-04-15 17:22:39
for the curd operation and functions
from sqlalchemy.orm import Session # for typing
from . import scheme # for using schemes like UserCreate, ItemCreate
from . import models # for using User and Item
def get_user(db: Session, user_id: int):
"""
description using db Session making query for users table to get
current user id
"""
return db.query(models.User).filter(models.User.id == user_id).first()
def get_user_by_email(db: Session, email: str):
"""
description using db Session making query for users table to get
current email
"""
return db.query(models.User).filter(models.User.email == email).first()
def get_users(db: Session, skip: int = 0, limit: int = 100):
"""
description using db Session get all user with limit 100 user
"""
return (
db.query(models.User)
.order_by(models.User.id)
.offset(skip)
.limit(limit)
.all()
)
def create_user(db: Session, user: scheme.UserCreate):
"""
description: create user and hashed the user password ;)
"""
fake_hashed = str(user.password) + "hashed value"
## create a user instance from User and adding password and email
db_user = models.User(
email=user.email, hash_password=fake_hashed
)
db.add(db_user) ## adding to db
db.commit() ## commit the changes
## reload the attributes of the db_user object from the database.
db.refresh(db_user)
# return db_user
return db_user
def get_items(db: Session, skip: int = 0, limit: int = 100):
"""
description: get all items i db with limit 100
"""
return (
db.query(models.Item)
.order_by(models.Item.id)
.offset(skip)
.limit(limit)
.all()
)
def create_user_item(db: Session, item: scheme.ItemCreate, user_id: int):
"""
creating item for user and adding the relationship between user and item
"""
db_item = models.Item(**item.dict(), owner_id = user_id)
db.add(db_item)
db.commit()
db.refresh(db_item)
return db_item
continue:./main.md
before:./models.md