FilesExpand file tree

 master

/

tutorial001.py

Copy path

Blame

More file actions

Blame

More file actions

Latest commit

 

History

History

History

65 lines (44 loc) · 1.38 KB

 master

/

tutorial001.py

Copy path

Top

File metadata and controls

• Code
• Blame

65 lines (44 loc) · 1.38 KB

Raw

Copy raw file

Download raw file

Open symbols panel

Edit and raw actions

1

2

3

4

5

6

7

8

9

10

11

12

13

14

15

16

17

18

19

20

21

22

23

24

25

26

27

28

29

30

31

32

33

34

35

36

37

38

39

40

41

42

43

44

45

46

47

48

49

50

51

52

53

54

55

56

57

58

59

60

61

62

63

64

65

from typing import List

import databases

import sqlalchemy

from fastapi import FastAPI

from pydantic import BaseModel

# SQLAlchemy specific code, as with any other app

DATABASE_URL = "sqlite:///./test.db"

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

database = databases.Database(DATABASE_URL)

metadata = sqlalchemy.MetaData()

notes = sqlalchemy.Table(

"notes",

metadata,

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

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

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

)

engine = sqlalchemy.create_engine(

DATABASE_URL, connect_args={"check_same_thread": False}

)

metadata.create_all(engine)

class NoteIn(BaseModel):

text: str

completed: bool

class Note(BaseModel):

id: int

text: str

completed: bool

app = FastAPI()

@app.on_event("startup")

async def startup():

await database.connect()

@app.on_event("shutdown")

async def shutdown():

await database.disconnect()

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

async def read_notes():

query = notes.select()

return await database.fetch_all(query)

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

async def create_note(note: NoteIn):

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

last_record_id = await database.execute(query)

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