-
Notifications
You must be signed in to change notification settings - Fork 27
/
main.py
84 lines (54 loc) · 1.72 KB
/
main.py
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
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
from fastapi import FastAPI
from fastapi.responses import PlainTextResponse, HTMLResponse
from pydantic import BaseModel
import sqlite3
app = FastAPI()
@app.get("/")
async def root():
return {"message": "Hello World"}
@app.get("/my_endpoint/")
async def my_new_endpoint():
return {"message": "Hello World. I have created this endpoint myself"}
@app.get("/plaintext/", response_class=PlainTextResponse)
async def plain_text():
return "Hello World\nI have created this endpoint myself"
@app.get("/html/", response_class=HTMLResponse)
async def html():
return "<h1>Hello World</h1><p>I have created this endpoint myself</p>"
@app.get("/items/{item_id}")
async def read_item(item_id):
return {"item_id": item_id}
@app.get("/greetings")
async def read_item(name: str = "Alice"):
return f"The name is {name}"
class Item(BaseModel):
name: str
description: str | None = None
price: float
tax: float | None = None
@app.post("/items/")
async def create_item(item: Item):
return f"Item created: {item}"
conn = sqlite3.connect('users.db')
c = conn.cursor()
c.execute('''CREATE TABLE IF NOT EXISTS users (id INTEGER PRIMARY KEY, name TEXT, age INTEGER)''')
conn.commit()
conn.close()
@app.get("/users")
async def get_users():
conn = sqlite3.connect('users.db')
c = conn.cursor()
c.execute('SELECT * FROM users')
users_list = c.fetchall()
return users_list
class User(BaseModel):
name: str
age: int
@app.post("/users")
async def create_user(user: User):
conn = sqlite3.connect('users.db')
c = conn.cursor()
c.execute('''INSERT INTO users (name, age) VALUES (?, ?)''', (user.name, user.age))
conn.commit()
conn.close()
return f"Created new user: {user}"