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

1from typing import List 5 ctx1abcde

2 

3from fastapi import Depends, FastAPI, HTTPException 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# Dependency 

15def get_db(): 5 ctx1abcde

16 db = SessionLocal() 30 ctx1fgpzqAhirBsCjktDuElmvFwGnoxHyI

17 try: 30 ctx1fgpzqAhirBsCjktDuElmvFwGnoxHyI

18 yield db 30 ctx1fgpzqAhirBsCjktDuElmvFwGnoxHyI

19 finally: 

20 db.close() 30 ctx1fgpzqAhirBsCjktDuElmvFwGnoxHyI

21 

22 

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

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

25 db_user = crud.get_user_by_email(db, email=user.email) 10 ctx1gJiKkLmMoN

26 if db_user: 10 ctx1gJiKkLmMoN

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

28 return crud.create_user(db=db, user=user) 10 ctx1gJiKkLmMoN

29 

30 

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

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

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

34 return users 5 ctx1zBDFH

35 

36 

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

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

39 db_user = crud.get_user(db, user_id=user_id) 20 ctx1fpqJhrsKjtuLlvwMnxyN

40 if db_user is None: 20 ctx1fpqJhrsKjtuLlvwMnxyN

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

42 return db_user 15 ctx1fpJhrKjtLlvMnxN

43 

44 

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

46def create_item_for_user( 5 ctx1abcde

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

48): 

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

50 

51 

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

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

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

55 return items 5 ctx1ACEGI