Last active
July 25, 2021 09:32
-
-
Save onecrayon/dd4803a5099061fa48d52f2d4bc2396b to your computer and use it in GitHub Desktop.
async/await FastAPI with SQLAlchemy test
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
#!/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) |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
[[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" |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
{ | |
"_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" | |
} | |
} | |
} |
It's done! It's on PyPI officially :)
See, https://docs.sqlalchemy.org/en/14/orm/extensions/asyncio.html
with the caveats outlined in the same page:
https://docs.sqlalchemy.org/en/14/orm/extensions/asyncio.html#preventing-implicit-io-when-using-asyncsession
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
Any news when they about to release this version?