Coverage for docs_src/sql_databases/sql_app/alt_main.py: 100%

1from typing import List 5 ctx1abcde

2 

3from fastapi import Depends, FastAPI, HTTPException, Request, Response 5 ctx1abcde

4from sqlalchemy.orm import Session 5 ctx1abcde

5 

6from . import crud, models, schemas 5 ctx1abcde

7from .database import SessionLocal, engine 5 ctx1abcde

8 

9models.Base.metadata.create_all(bind=engine) 5 ctx1abcde

10 

11app = FastAPI() 5 ctx1abcde

12 

13 

14@app.middleware("http") 5 ctx1abcde

15async def db_session_middleware(request: Request, call_next): 5 ctx1abcde

16 response = Response("Internal server error", status_code=500) 35 ctx1fgpzqJAhirBsKCjktDuLElmvFwMGnoxHyNI

17 try: 35 ctx1fgpzqJAhirBsKCjktDuLElmvFwMGnoxHyNI

18 request.state.db = SessionLocal() 35 ctx1fgpzqJAhirBsKCjktDuLElmvFwMGnoxHyNI

19 response = await call_next(request) 35 ctx1fgpzqJAhirBsKCjktDuLElmvFwMGnoxHyNI

20 finally: 

21 request.state.db.close() 35 ctx1fgpzqJAhirBsKCjktDuLElmvFwMGnoxHyNI

22 return response 35 ctx1fgpzqJAhirBsKCjktDuLElmvFwMGnoxHyNI

23 

24 

25# Dependency 

26def get_db(request: Request): 5 ctx1abcde

27 return request.state.db 30 ctx1fgpzqAhirBsCjktDuElmvFwGnoxHyI

28 

29 

30@app.post("/users/", response_model=schemas.User) 5 ctx1abcde

31def create_user(user: schemas.UserCreate, db: Session = Depends(get_db)): 5 ctx1abcde

32 db_user = crud.get_user_by_email(db, email=user.email) 5 ctx1gikmo

33 if db_user: 5 ctx1gikmo

34 raise HTTPException(status_code=400, detail="Email already registered") 5 ctx1gikmo

35 return crud.create_user(db=db, user=user) 5 ctx1gikmo

36 

37 

38@app.get("/users/", response_model=List[schemas.User]) 5 ctx1abcde

39def read_users(skip: int = 0, limit: int = 100, db: Session = Depends(get_db)): 5 ctx1abcde

40 users = crud.get_users(db, skip=skip, limit=limit) 5 ctx1zBDFH

41 return users 5 ctx1zBDFH

42 

43 

44@app.get("/users/{user_id}", response_model=schemas.User) 5 ctx1abcde

45def read_user(user_id: int, db: Session = Depends(get_db)): 5 ctx1abcde

46 db_user = crud.get_user(db, user_id=user_id) 15 ctx1fpqhrsjtulvwnxy

47 if db_user is None: 15 ctx1fpqhrsjtulvwnxy

48 raise HTTPException(status_code=404, detail="User not found") 5 ctx1qsuwy

49 return db_user 10 ctx1fphrjtlvnx

50 

51 

52@app.post("/users/{user_id}/items/", response_model=schemas.Item) 5 ctx1abcde

53def create_item_for_user( 5 ctx1abcde

54 user_id: int, item: schemas.ItemCreate, db: Session = Depends(get_db) 

55): 

56 return crud.create_user_item(db=db, item=item, user_id=user_id) 5 ctx1fhjln

57 

58 

59@app.get("/items/", response_model=List[schemas.Item]) 5 ctx1abcde

60def read_items(skip: int = 0, limit: int = 100, db: Session = Depends(get_db)): 5 ctx1abcde

61 items = crud.get_items(db, skip=skip, limit=limit) 5 ctx1ACEGI

62 return items 5 ctx1ACEGI