Initiale DAteien hinzugefuegt
This commit is contained in:
+36
@@ -0,0 +1,36 @@
|
||||
from sqlalchemy.orm import Session
|
||||
|
||||
from . import models,schemas
|
||||
|
||||
def get_user(db: Session, user_id: int):
|
||||
return db.query(models.User).filter(models.User.id == user_id).first()
|
||||
|
||||
def get_user_by_email(db: Session, email: str):
|
||||
return db.query(models.User).filter(models.User.email == email).first()
|
||||
|
||||
def get_users(db: Session, skip:int=0, limit:int=100):
|
||||
# return db.query(models.User).offset(skip).limit(limit).all()
|
||||
return db.query(models.User).offset(skip).limit(limit).all()
|
||||
|
||||
def create_user(db: Session, user:schemas.UserCreate):
|
||||
db_user = models.User(email=user.email,
|
||||
name=user.name)
|
||||
db.add(db_user)
|
||||
db.commit()
|
||||
db.refresh(db_user)
|
||||
return db_user
|
||||
|
||||
def get_todos(db: Session, skip:int=0, limit: int=100):
|
||||
return db.query(models.Todo).offset(skip).limit(limit).all()
|
||||
|
||||
def create_user_todo(db:Session, todo:schemas.TodoCreate, user_id : int):
|
||||
db_todo = models.Todo(**todo.model_dump(),owner_id=user_id )
|
||||
db.add(db_todo)
|
||||
db.commit()
|
||||
db.refresh(db_todo)
|
||||
return db_todo
|
||||
|
||||
# NOTE :
|
||||
# - add that instance object to your database session.
|
||||
# - commit the changes to the database (so that they are saved).
|
||||
# - refresh your instance (so that it contains any new data from the database, like the generated ID).
|
||||
@@ -0,0 +1,12 @@
|
||||
from sqlalchemy import create_engine
|
||||
from sqlalchemy.ext.declarative import declarative_base
|
||||
from sqlalchemy.orm import sessionmaker
|
||||
from dotenv import load_dotenv
|
||||
|
||||
load_dotenv()
|
||||
import os
|
||||
DB_URL = DB_URL = os.getenv("DB_URL")
|
||||
engine = create_engine(DB_URL,echo=True)
|
||||
SessionLocal = sessionmaker(autocommit=False,autoflush=False, bind=engine)
|
||||
|
||||
Base = declarative_base()
|
||||
+45
@@ -0,0 +1,45 @@
|
||||
from fastapi import FastAPI, Depends, HTTPException
|
||||
from sqlalchemy.orm import Session
|
||||
|
||||
from . import crud,models, schemas
|
||||
from .database import SessionLocal, engine
|
||||
|
||||
models.Base.metadata.create_all(bind=engine)
|
||||
|
||||
app = FastAPI()
|
||||
|
||||
#Dependency
|
||||
def get_db():
|
||||
db = SessionLocal()
|
||||
try :
|
||||
yield db
|
||||
finally:
|
||||
db.close()
|
||||
|
||||
@app.post("/users/",response_model=schemas.User)
|
||||
def post_user(user:schemas.UserCreate, db:Session=Depends(get_db)):
|
||||
db_user = crud.get_user_by_email(db, email=user.email)
|
||||
if db_user:
|
||||
raise HTTPException(status_code=400, detail="Email already registered")
|
||||
return crud.create_user(db=db,user=user)
|
||||
|
||||
@app.get("/users/", response_model=list[schemas.User])
|
||||
def get_users(skip:int=0, limit:int=0, db:Session=Depends(get_db)):
|
||||
users = crud.get_users(db,skip=skip,limit=limit)
|
||||
return users
|
||||
|
||||
@app.get("/users/{user_id}/",response_model=schemas.User)
|
||||
def get_user(user_id:int, db:Session=Depends(get_db)):
|
||||
db_user = crud.get_user(db,user_id =user_id )
|
||||
if db_user is None:
|
||||
raise HTTPException(status_code=404, detail="User not found")
|
||||
return db_user
|
||||
|
||||
@app.post("/users/{user_id}/todos/",response_model=schemas.Todo)
|
||||
def post_todo_for_user(user_id:int, todo:schemas.TodoCreate, db:Session=Depends(get_db)):
|
||||
return crud.create_user_todo(db=db,user_id=user_id, todo=todo)
|
||||
|
||||
@app.get("/todos/", response_model=list[schemas.Todo])
|
||||
def get_todos(skip:int=0,limit:int=100,db:Session=Depends(get_db)):
|
||||
todos = crud.get_todos(db,skip=skip,limit=limit)
|
||||
return todos
|
||||
@@ -0,0 +1,21 @@
|
||||
from sqlalchemy import Boolean, Column, ForeignKey, Integer, String
|
||||
from sqlalchemy.orm import relationship
|
||||
|
||||
from .database import Base
|
||||
|
||||
class User(Base):
|
||||
__tablename__ = "users"
|
||||
id = Column(Integer,primary_key=True,index=True)
|
||||
name = Column(String(255),index=True)
|
||||
email = Column(String(255), unique=True, index=True)
|
||||
todos = relationship("Todo",back_populates="owner")
|
||||
is_active = Column(Boolean,default=False)
|
||||
|
||||
class Todo(Base):
|
||||
__tablename__ = "todos"
|
||||
id = Column(Integer, primary_key=True, index=True)
|
||||
title = Column(String(255), index=True)
|
||||
description = Column(String(255), index=True)
|
||||
owner_id = Column(Integer, ForeignKey("users.id"))
|
||||
|
||||
owner = relationship("User",back_populates="todos")
|
||||
@@ -0,0 +1,30 @@
|
||||
from pydantic import BaseModel
|
||||
|
||||
class TodoBase(BaseModel):
|
||||
title : str
|
||||
description : str | None = None
|
||||
|
||||
class TodoCreate(TodoBase):
|
||||
pass
|
||||
|
||||
class Todo(TodoBase):
|
||||
id : int
|
||||
owner_id : int
|
||||
|
||||
class Config:
|
||||
orm_mode = True
|
||||
|
||||
class UserBase(BaseModel):
|
||||
email: str
|
||||
name: str
|
||||
|
||||
class UserCreate(UserBase):
|
||||
pass
|
||||
|
||||
class User(UserBase):
|
||||
id : int
|
||||
is_active : bool
|
||||
todos : list[Todo] = []
|
||||
|
||||
class Config:
|
||||
orm_model = True
|
||||
Reference in New Issue
Block a user