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
« prev ^ index » next coverage.py v7.6.1, created at 2024-08-08 03:53 +0000
1from typing import List 1abcde
3import databases 1abcde
4import sqlalchemy 1abcde
5from fastapi import FastAPI 1abcde
6from pydantic import BaseModel 1abcde
8# SQLAlchemy specific code, as with any other app
9DATABASE_URL = "sqlite:///./test.db" 1abcde
10# DATABASE_URL = "postgresql://user:password@postgresserver/db"
12database = databases.Database(DATABASE_URL) 1abcde
14metadata = sqlalchemy.MetaData() 1abcde
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)
25engine = sqlalchemy.create_engine( 1abcde
26 DATABASE_URL, connect_args={"check_same_thread": False}
27)
28metadata.create_all(engine) 1abcde
31class NoteIn(BaseModel): 1abcde
32 text: str 1abcde
33 completed: bool 1abcde
36class Note(BaseModel): 1abcde
37 id: int 1abcde
38 text: str 1abcde
39 completed: bool 1abcde
42app = FastAPI() 1abcde
45@app.on_event("startup") 1abcde
46async def startup(): 1abcde
47 await database.connect() 1abcde
50@app.on_event("shutdown") 1abcde
51async def shutdown(): 1abcde
52 await database.disconnect() 1abcde
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
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