Skip to content

Instantly share code, notes, and snippets.

@onecrayon
Last active July 25, 2021 09:32
Show Gist options
  • Save onecrayon/dd4803a5099061fa48d52f2d4bc2396b to your computer and use it in GitHub Desktop.
Save onecrayon/dd4803a5099061fa48d52f2d4bc2396b to your computer and use it in GitHub Desktop.
async/await FastAPI with SQLAlchemy test
#!/usr/bin/env python3
"""
1. Install Python 3.7 and `pipenv`: https://docs.pipenv.org/en/latest/
2. `pipenv install`
3. `pipenv shell`
4. `./main.py`
5. In your browser, visit http://localhost:8000/parents
See comments in lines 92-108 for details on where I'm struggling.
"""
import logging
from typing import List
from pprint import pprint
import uvicorn
from databases import Database
from fastapi import FastAPI
from pydantic import BaseModel
import sqlalchemy as sa
from sqlalchemy.engine import Engine
from sqlalchemy.orm import relationship, joinedload
from sqlalchemy.orm.query import Query
from sqlalchemy.ext.declarative import declarative_base
logger = logging.getLogger()
logger.setLevel(logging.DEBUG)
AlchemyBase = declarative_base()
DATABASE_URL = "sqlite:///./test.db"
db = Database(DATABASE_URL)
"""SQLAlchemy model classes"""
class Parent(AlchemyBase):
__tablename__ = 'parents'
id = sa.Column(sa.Integer, primary_key=True)
name = sa.Column(sa.String)
children = relationship('Child', back_populates='parent')
class Child(AlchemyBase):
__tablename__ = 'children'
id = sa.Column(sa.Integer, primary_key=True)
name = sa.Column(sa.String)
parent_id = sa.Column(sa.Integer, sa.ForeignKey('parents.id'))
parent = relationship('Parent', back_populates='children')
"""Pydantic FastAPI type classes"""
class ChildOut(BaseModel):
id: int
name: str
class ParentOut(BaseModel):
id: int
name: str
children: List[ChildOut] = []
"""Load app and initialize database"""
app = FastAPI()
### Bootstrap the database; this is only necessary to create the tables and data for this example
# I would manage the database with migrations rather than creating an engine normally
engine: Engine = sa.create_engine(
DATABASE_URL, connect_args={"check_same_thread": False}
)
AlchemyBase.metadata.create_all(bind=engine)
engine.execute("INSERT OR REPLACE INTO parents (id, name) VALUES (1, 'First Parent'), (2, 'Second Parent'), (3, 'Third Parent')")
engine.execute("INSERT OR REPLACE INTO children (id, name, parent_id) VALUES (1, 'Child A', 1), (2, 'Child B', 1), (3, 'Child C', 2)")
### End example bootstrapping code
"""Define app events for managing database lifecycle"""
@app.on_event("startup")
async def startup():
await db.database.connect()
@app.on_event("shutdown")
async def shutdown():
await db.database.disconnect()
"""Define app routes for fetching data"""
@app.get('/parents', response_model=List[ParentOut])
async def read_parents():
query = Query(Parent).options(joinedload('children'))
# This logic is how SQLAlchemy handles things internally; similarly, I could just do this:
# statement = query.with_labels().statement
# results = await db.fetch_all(statement)
# Best I can tell, I'll need access to the context to correctly map stuff (hence this code)
query_context = query._compile_context()
query_context.statement.use_labels = True
logger.info(f'QUERY: {query_context.statement}')
results = await db.fetch_all(statement)
# For logging purposes...
results_as_dicts = [
{column: value for column, value in rowproxy.items()} for rowproxy in results
]
logger.info(f'RESULTS: {pprint(results_as_dicts)}')
# At this point I have a query, a context, and a result set, and I need to use those things to
# map the results into a dict that I can return (at which point Pydantic will convert the dict
# into a list of ParentOut objects)
return []
"""Easy server running for debugging"""
if __name__ == "__main__":
uvicorn.run(app, host="localhost", port=8000)
[[source]]
name = "pypi"
url = "https://pypi.org/simple"
verify_ssl = true
[dev-packages]
pylint = "*"
[packages]
sqlalchemy = "*"
databases = {extras = ["sqlite"],version = "*"}
fastapi = "*"
uvicorn = "*"
[requires]
python_version = "3.7"
{
"_meta": {
"hash": {
"sha256": "c2f63450a89805ec91a43a66ea7ef08c4e73e032a282a9ffba1d36ca7c827324"
},
"pipfile-spec": 6,
"requires": {
"python_version": "3.7"
},
"sources": [
{
"name": "pypi",
"url": "https://pypi.org/simple",
"verify_ssl": true
}
]
},
"default": {
"aiosqlite": {
"hashes": [
"sha256:ad84fbd7516ca7065d799504fc41d6845c938e5306d1b7dd960caaeda12e22a9"
],
"version": "==0.10.0"
},
"click": {
"hashes": [
"sha256:2335065e6395b9e67ca716de5f7526736bfa6ceead690adf616d925bdc622b13",
"sha256:5b94b49521f6456670fdb30cd82a4eca9412788a93fa6dd6df72c94d5a8ff2d7"
],
"version": "==7.0"
},
"databases": {
"extras": [
"sqlite"
],
"hashes": [
"sha256:1e3b21a237d8b8a8774da1237fa75e951e23bf8e943516df8fe2443f1968287f"
],
"index": "pypi",
"version": "==0.2.5"
},
"fastapi": {
"hashes": [
"sha256:53259a2b572e49ac2634fc643912c6a818623242705ebe2a060db4f51807a83e",
"sha256:7117cb8464b3525cd5c5b09ed44628425ca62ce65925f5076b52faf2e7078b5e"
],
"index": "pypi",
"version": "==0.31.0"
},
"h11": {
"hashes": [
"sha256:acca6a44cb52a32ab442b1779adf0875c443c689e9e028f8d831a3769f9c5208",
"sha256:f2b1ca39bfed357d1f19ac732913d5f9faa54a5062eca7d2ec3a916cfb7ae4c7"
],
"version": "==0.8.1"
},
"httptools": {
"hashes": [
"sha256:e00cbd7ba01ff748e494248183abc6e153f49181169d8a3d41bb49132ca01dfc"
],
"markers": "sys_platform != 'win32' and sys_platform != 'cygwin' and platform_python_implementation != 'pypy'",
"version": "==0.0.13"
},
"pydantic": {
"hashes": [
"sha256:098ffe36047a35f78dbc0d8f7c2d40da053e35158b6f4326c50d669c56e48ea7",
"sha256:549c5d05138b7e0f33bd24d4313783c90342d0595d824e02ceb562a992c78239",
"sha256:6901572483a00d53aaba9121c6261017334511c44864b00b36714b32cc7af712",
"sha256:70a5cbab8b52ac3e5f9660a8ad028581095e48abc4a0fc67d1b2bf69e8a51d03",
"sha256:9d34421072a7df679638e5896a4a15f48ffbbfb8a57eb5fac6c45806269d0dc0",
"sha256:e4968f16f448d7c3026b0197bd29e78b5f39a7d27dbe22d5b3f07fb8602a2dc1"
],
"version": "==0.29"
},
"sqlalchemy": {
"hashes": [
"sha256:c30925d60af95443458ebd7525daf791f55762b106049ae71e18f8dd58084c2f"
],
"index": "pypi",
"version": "==1.3.5"
},
"starlette": {
"hashes": [
"sha256:d313433ef5cc38e0a276b59688ca2b11b8f031c78808c1afdf9d55cb86f34590"
],
"version": "==0.12.0"
},
"uvicorn": {
"hashes": [
"sha256:5320b48629564fa914fe8120542608c91f98fcd2278e8454bda9db08467519d1"
],
"index": "pypi",
"version": "==0.8.4"
},
"uvloop": {
"hashes": [
"sha256:0fcd894f6fc3226a962ee7ad895c4f52e3f5c3c55098e21efb17c071849a0573",
"sha256:2f31de1742c059c96cb76b91c5275b22b22b965c886ee1fced093fa27dde9e64",
"sha256:459e4649fcd5ff719523de33964aa284898e55df62761e7773d088823ccbd3e0",
"sha256:67867aafd6e0bc2c30a079603a85d83b94f23c5593b3cc08ec7e58ac18bf48e5",
"sha256:8c200457e6847f28d8bb91c5e5039d301716f5f2fce25646f5fb3fd65eda4a26",
"sha256:958906b9ca39eb158414fbb7d6b8ef1b7aee4db5c8e8e5d00fcbb69a1ce9dca7",
"sha256:ac1dca3d8f3ef52806059e81042ee397ac939e5a86c8a3cea55d6b087db66115",
"sha256:b284c22d8938866318e3b9d178142b8be316c52d16fcfe1560685a686718a021",
"sha256:c48692bf4587ce281d641087658eca275a5ad3b63c78297bbded96570ae9ce8f",
"sha256:fefc3b2b947c99737c348887db2c32e539160dcbeb7af9aa6b53db7a283538fe"
],
"markers": "sys_platform != 'win32' and sys_platform != 'cygwin' and platform_python_implementation != 'pypy'",
"version": "==0.12.2"
},
"websockets": {
"hashes": [
"sha256:04b42a1b57096ffa5627d6a78ea1ff7fad3bc2c0331ffc17bc32a4024da7fea0",
"sha256:08e3c3e0535befa4f0c4443824496c03ecc25062debbcf895874f8a0b4c97c9f",
"sha256:10d89d4326045bf5e15e83e9867c85d686b612822e4d8f149cf4840aab5f46e0",
"sha256:232fac8a1978fc1dead4b1c2fa27c7756750fb393eb4ac52f6bc87ba7242b2fa",
"sha256:4bf4c8097440eff22bc78ec76fe2a865a6e658b6977a504679aaf08f02c121da",
"sha256:51642ea3a00772d1e48fb0c492f0d3ae3b6474f34d20eca005a83f8c9c06c561",
"sha256:55d86102282a636e195dad68aaaf85b81d0bef449d7e2ef2ff79ac450bb25d53",
"sha256:564d2675682bd497b59907d2205031acbf7d3fadf8c763b689b9ede20300b215",
"sha256:5d13bf5197a92149dc0badcc2b699267ff65a867029f465accfca8abab95f412",
"sha256:5eda665f6789edb9b57b57a159b9c55482cbe5b046d7db458948370554b16439",
"sha256:5edb2524d4032be4564c65dc4f9d01e79fe8fad5f966e5b552f4e5164fef0885",
"sha256:79691794288bc51e2a3b8de2bc0272ca8355d0b8503077ea57c0716e840ebaef",
"sha256:7fcc8681e9981b9b511cdee7c580d5b005f3bb86b65bde2188e04a29f1d63317",
"sha256:8e447e05ec88b1b408a4c9cde85aa6f4b04f06aa874b9f0b8e8319faf51b1fee",
"sha256:90ea6b3e7787620bb295a4ae050d2811c807d65b1486749414f78cfd6fb61489",
"sha256:9e13239952694b8b831088431d15f771beace10edfcf9ef230cefea14f18508f",
"sha256:d40f081187f7b54d7a99d8a5c782eaa4edc335a057aa54c85059272ed826dc09",
"sha256:e1df1a58ed2468c7b7ce9a2f9752a32ad08eac2bcd56318625c3647c2cd2da6f",
"sha256:e98d0cec437097f09c7834a11c69d79fe6241729b23f656cfc227e93294fc242",
"sha256:f8d59627702d2ff27cb495ca1abdea8bd8d581de425c56e93bff6517134e0a9b",
"sha256:fc30cdf2e949a2225b012a7911d1d031df3d23e99b7eda7dfc982dc4a860dae9"
],
"version": "==7.0"
}
},
"develop": {
"astroid": {
"hashes": [
"sha256:6560e1e1749f68c64a4b5dee4e091fce798d2f0d84ebe638cf0e0585a343acf4",
"sha256:b65db1bbaac9f9f4d190199bb8680af6f6f84fd3769a5ea883df8a91fe68b4c4"
],
"version": "==2.2.5"
},
"isort": {
"hashes": [
"sha256:54da7e92468955c4fceacd0c86bd0ec997b0e1ee80d97f67c35a78b719dccab1",
"sha256:6e811fcb295968434526407adb8796944f1988c5b65e8139058f2014cbe100fd"
],
"version": "==4.3.21"
},
"lazy-object-proxy": {
"hashes": [
"sha256:159a745e61422217881c4de71f9eafd9d703b93af95618635849fe469a283661",
"sha256:23f63c0821cc96a23332e45dfaa83266feff8adc72b9bcaef86c202af765244f",
"sha256:3b11be575475db2e8a6e11215f5aa95b9ec14de658628776e10d96fa0b4dac13",
"sha256:3f447aff8bc61ca8b42b73304f6a44fa0d915487de144652816f950a3f1ab821",
"sha256:4ba73f6089cd9b9478bc0a4fa807b47dbdb8fad1d8f31a0f0a5dbf26a4527a71",
"sha256:4f53eadd9932055eac465bd3ca1bd610e4d7141e1278012bd1f28646aebc1d0e",
"sha256:64483bd7154580158ea90de5b8e5e6fc29a16a9b4db24f10193f0c1ae3f9d1ea",
"sha256:6f72d42b0d04bfee2397aa1862262654b56922c20a9bb66bb76b6f0e5e4f9229",
"sha256:7c7f1ec07b227bdc561299fa2328e85000f90179a2f44ea30579d38e037cb3d4",
"sha256:7c8b1ba1e15c10b13cad4171cfa77f5bb5ec2580abc5a353907780805ebe158e",
"sha256:8559b94b823f85342e10d3d9ca4ba5478168e1ac5658a8a2f18c991ba9c52c20",
"sha256:a262c7dfb046f00e12a2bdd1bafaed2408114a89ac414b0af8755c696eb3fc16",
"sha256:acce4e3267610c4fdb6632b3886fe3f2f7dd641158a843cf6b6a68e4ce81477b",
"sha256:be089bb6b83fac7f29d357b2dc4cf2b8eb8d98fe9d9ff89f9ea6012970a853c7",
"sha256:bfab710d859c779f273cc48fb86af38d6e9210f38287df0069a63e40b45a2f5c",
"sha256:c10d29019927301d524a22ced72706380de7cfc50f767217485a912b4c8bd82a",
"sha256:dd6e2b598849b3d7aee2295ac765a578879830fb8966f70be8cd472e6069932e",
"sha256:e408f1eacc0a68fed0c08da45f31d0ebb38079f043328dce69ff133b95c29dc1"
],
"version": "==1.4.1"
},
"mccabe": {
"hashes": [
"sha256:ab8a6258860da4b6677da4bd2fe5dc2c659cff31b3ee4f7f5d64e79735b80d42",
"sha256:dd8d182285a0fe56bace7f45b5e7d1a6ebcbf524e8f3bd87eb0f125271b8831f"
],
"version": "==0.6.1"
},
"pylint": {
"hashes": [
"sha256:5d77031694a5fb97ea95e828c8d10fc770a1df6eb3906067aaed42201a8a6a09",
"sha256:723e3db49555abaf9bf79dc474c6b9e2935ad82230b10c1138a71ea41ac0fff1"
],
"index": "pypi",
"version": "==2.3.1"
},
"six": {
"hashes": [
"sha256:3350809f0555b11f552448330d0b52d5f24c91a322ea4a15ef22629740f3761c",
"sha256:d16a0141ec1a18405cd4ce8b4613101da75da0e9a7aec5bdd4fa804d0e0eba73"
],
"version": "==1.12.0"
},
"typed-ast": {
"hashes": [
"sha256:18511a0b3e7922276346bcb47e2ef9f38fb90fd31cb9223eed42c85d1312344e",
"sha256:262c247a82d005e43b5b7f69aff746370538e176131c32dda9cb0f324d27141e",
"sha256:2b907eb046d049bcd9892e3076c7a6456c93a25bebfe554e931620c90e6a25b0",
"sha256:354c16e5babd09f5cb0ee000d54cfa38401d8b8891eefa878ac772f827181a3c",
"sha256:4e0b70c6fc4d010f8107726af5fd37921b666f5b31d9331f0bd24ad9a088e631",
"sha256:630968c5cdee51a11c05a30453f8cd65e0cc1d2ad0d9192819df9978984529f4",
"sha256:66480f95b8167c9c5c5c87f32cf437d585937970f3fc24386f313a4c97b44e34",
"sha256:71211d26ffd12d63a83e079ff258ac9d56a1376a25bc80b1cdcdf601b855b90b",
"sha256:95bd11af7eafc16e829af2d3df510cecfd4387f6453355188342c3e79a2ec87a",
"sha256:bc6c7d3fa1325a0c6613512a093bc2a2a15aeec350451cbdf9e1d4bffe3e3233",
"sha256:cc34a6f5b426748a507dd5d1de4c1978f2eb5626d51326e43280941206c209e1",
"sha256:d755f03c1e4a51e9b24d899561fec4ccaf51f210d52abdf8c07ee2849b212a36",
"sha256:d7c45933b1bdfaf9f36c579671fec15d25b06c8398f113dab64c18ed1adda01d",
"sha256:d896919306dd0aa22d0132f62a1b78d11aaf4c9fc5b3410d3c666b818191630a",
"sha256:ffde2fbfad571af120fcbfbbc61c72469e72f550d676c3342492a9dfdefb8f12"
],
"markers": "implementation_name == 'cpython'",
"version": "==1.4.0"
},
"wrapt": {
"hashes": [
"sha256:565a021fd19419476b9362b05eeaa094178de64f8361e44468f9e9d7843901e1",
"sha256:cde1b271fe534b474479d52f9d143499458253395b7847f7bab57872dfe27bd8"
],
"version": "==1.11.2"
}
}
}
@janaSunrise
Copy link

It's done! It's on PyPI officially :)

Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment