108 lines
3.2 KiB
Python
108 lines
3.2 KiB
Python
from fastapi import FastAPI, HTTPException
|
|
from typing import List
|
|
from sqlmodel import SQLModel, Session, select
|
|
from sqlalchemy.orm import selectinload, joinedload, subqueryload
|
|
from models import User, Post
|
|
from models import *
|
|
from database import engine
|
|
from dependencies import get_session
|
|
from fastapi import Depends
|
|
from service import get_users_with_recent_posts
|
|
|
|
app = FastAPI()
|
|
|
|
|
|
@app.on_event("startup")
|
|
def on_startup():
|
|
SQLModel.metadata.create_all(engine)
|
|
|
|
|
|
@app.get("/users", response_model=List[User])
|
|
def get_users_with_posts(session: Session = Depends(get_session)):
|
|
return User.get_all(session)
|
|
|
|
|
|
@app.get("/users/recent", response_model=List[UserWithRecentPosts])
|
|
def get_users_recent(session: Session = Depends(get_session)):
|
|
return get_users_with_recent_posts(session)
|
|
|
|
|
|
@app.get("/users/{user_id}/posts", response_model=List[PostReadWithUser])
|
|
def get_posts_by_user(user_id: int, session: Session = Depends(get_session)):
|
|
with Session(engine) as session:
|
|
statement = (
|
|
select(Post).where(Post.user_id == user_id).options(selectinload(Post.user))
|
|
)
|
|
|
|
posts = session.exec(statement).all()
|
|
|
|
if not posts:
|
|
raise HTTPException(status_code=404, detail="No posts found for user")
|
|
|
|
return posts
|
|
# return [PostReadWithUser.model_validate(post) for post in posts]
|
|
|
|
|
|
@app.get("/nplusone", response_model=List[PostReadWithUser])
|
|
def get_nplusone(session: Session = Depends(get_session)):
|
|
user_id = 1
|
|
with Session(engine) as session:
|
|
statement = select(Post).where(Post.user_id == user_id)
|
|
|
|
posts = session.exec(statement).all()
|
|
|
|
for post in posts:
|
|
session.refresh(post.user)
|
|
|
|
if not posts:
|
|
raise HTTPException(status_code=404, detail="No posts found for user")
|
|
|
|
return posts
|
|
|
|
|
|
@app.get("/selectinload", response_model=List[PostReadWithUser])
|
|
def get_selectinload(session: Session = Depends(get_session)):
|
|
user_id = 1
|
|
with Session(engine) as session:
|
|
statement = (
|
|
select(Post).where(Post.user_id == user_id).options(selectinload(Post.user))
|
|
)
|
|
|
|
posts = session.exec(statement).all()
|
|
|
|
if not posts:
|
|
raise HTTPException(status_code=404, detail="No posts found for user")
|
|
|
|
return posts
|
|
|
|
|
|
@app.get("/joinedload", response_model=List[PostReadWithUser])
|
|
def get_joinedload(session: Session = Depends(get_session)):
|
|
user_id = 1
|
|
with Session(engine) as session:
|
|
statement = (
|
|
select(Post).where(Post.user_id == user_id).options(joinedload(Post.user))
|
|
)
|
|
|
|
posts = session.exec(statement).all()
|
|
|
|
if not posts:
|
|
raise HTTPException(status_code=404, detail="No posts found for user")
|
|
|
|
return posts
|
|
|
|
|
|
@app.get("/subqueryload", response_model=List[PostReadWithUser])
|
|
def get_subqueryload(session: Session = Depends(get_session)):
|
|
user_id = 1
|
|
with Session(engine) as session:
|
|
statement = (
|
|
select(Post).where(Post.user_id == user_id).options(subqueryload(Post.user))
|
|
)
|
|
|
|
posts = session.exec(statement).all()
|
|
|
|
if not posts:
|
|
raise HTTPException(status_code=404, detail="No posts found for user")
|
|
|
|
return posts
|