Coverage for docs_src/async_sql_databases/tutorial001.py: 100%

34 statements  

« prev     ^ index     » next       coverage.py v7.6.1, created at 2024-08-08 03:53 +0000

1from typing import List 1abcde

2 

3import databases 1abcde

4import sqlalchemy 1abcde

5from fastapi import FastAPI 1abcde

6from pydantic import BaseModel 1abcde

7 

8# SQLAlchemy specific code, as with any other app 

9DATABASE_URL = "sqlite:///./test.db" 1abcde

10# DATABASE_URL = "postgresql://user:password@postgresserver/db" 

11 

12database = databases.Database(DATABASE_URL) 1abcde

13 

14metadata = sqlalchemy.MetaData() 1abcde

15 

16notes = sqlalchemy.Table( 1abcde

17 "notes", 

18 metadata, 

19 sqlalchemy.Column("id", sqlalchemy.Integer, primary_key=True), 

20 sqlalchemy.Column("text", sqlalchemy.String), 

21 sqlalchemy.Column("completed", sqlalchemy.Boolean), 

22) 

23 

24 

25engine = sqlalchemy.create_engine( 1abcde

26 DATABASE_URL, connect_args={"check_same_thread": False} 

27) 

28metadata.create_all(engine) 1abcde

29 

30 

31class NoteIn(BaseModel): 1abcde

32 text: str 1abcde

33 completed: bool 1abcde

34 

35 

36class Note(BaseModel): 1abcde

37 id: int 1abcde

38 text: str 1abcde

39 completed: bool 1abcde

40 

41 

42app = FastAPI() 1abcde

43 

44 

45@app.on_event("startup") 1abcde

46async def startup(): 1abcde

47 await database.connect() 1abcde

48 

49 

50@app.on_event("shutdown") 1abcde

51async def shutdown(): 1abcde

52 await database.disconnect() 1abcde

53 

54 

55@app.get("/notes/", response_model=List[Note]) 1abcde

56async def read_notes(): 1abcde

57 query = notes.select() 1abcde

58 return await database.fetch_all(query) 1abcde

59 

60 

61@app.post("/notes/", response_model=Note) 1abcde

62async def create_note(note: NoteIn): 1abcde

63 query = notes.insert().values(text=note.text, completed=note.completed) 1abcde

64 last_record_id = await database.execute(query) 1abcde

65 return {**note.dict(), "id": last_record_id} 1abcde