-
Notifications
You must be signed in to change notification settings - Fork 0
/
main.py
63 lines (50 loc) · 1.82 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
from typing import Union
from fastapi import FastAPI, Request
from sqlalchemy import create_engine, text, insert, Column, Integer, String, Table, MetaData
from sqlalchemy.ext.declarative import declarative_base
from pydantic import BaseModel
from pydantic_sqlalchemy import sqlalchemy_to_pydantic
Base = declarative_base()
engine = create_engine('postgresql+psycopg2://max2:pswd@localhost/pgdb',future=True)
Base.metadata.create_all(engine)
metadata = MetaData()
quote_table = Table('Quotes',
metadata,
Column('Id', Integer, primary_key=True),
Column('Quote', String),
Column('Author', String)
)
class QuoteObject(Base):
__tablename__ = "Quotes"
id = Column(Integer, primary_key=True)
quote = Column(String)
author = Column(String)
PydanticQuote = sqlalchemy_to_pydantic(QuoteObject)
def getQuote(id):
with engine.connect() as conn:
result = conn.execution_options(stream_results=True).execute(text("SELECT Id, Quote, Author FROM Quotes WHERE Id = " + str(id)))
row = result.fetchone()
return row
def insertQuote(q: QuoteObject):
stmt = insert(quote_table).values(Id=q.id, Quote=q.quote, Author=q.author)
print(stmt)
with engine.connect() as conn:
result = conn.execute(stmt)
conn.commit()
print('Inserted Primary Key: %s' % str(result.inserted_primary_key))
app = FastAPI()
@app.get("/")
def read_root():
return {"Hello": "World"}
@app.get("/items/{item_id}")
def read_item(item_id: int):
quote = PydanticQuote.from_orm(getQuote(item_id))
print(quote)
return quote
@app.post("/items/")
async def create_item(request: Request):
req = await request.json()
quote = QuoteObject(**req)
print(quote.author)
insertQuote(quote)
return quote