Building a Database REST API with Python
In this article, I will be walking through building a REST API for a Database, and in as little time as possible.
Constructing an API quickly may be important, but so is the performance of the API. For this reason, the API will be built using the FastAPI combined with with uvicorn, easyrpc and aiopyql for handling requests, database access and caching.
For this article, python 3.7 or greater can be used, which this is the min version required by the above packages.
Create and activate python environment.
$ virtualenv -p python3.7 db-api
$ source db-api/bin/activateInstall our dependencies
(db-api)$ pip install easyrpc aiopyqlThe API will be structured with two layers:
- Database Access Layer: Database access and caching
- API Access Layer: Handle client API requests
Often this would be coupled into a single layer, but the benefits are explained below.
Database Layer
server = FastAPI()The FastAPI() instance server reference is our primary URL router for Client Requests in the API Layer, and a Websocket Router for EasyRpcServer in the Database Layer.
@server.on_event('startup')FastAPI on_event provides two useful hooks for performing setup / tear-down when the application starts. on_event(startup|shutdown)
# Rpc Server
rpc_server = await EasyRpcServer.create(
server,
'/ws/database',
server_secret='abcd1234'
)EasyRpcServer.create is a class factory method which returns a EasyRpcServer() instance: using our FastAPI server, web-socket path(where the ASGI server will accept Websocket negotiation), server_secret( the password that a EasyRpcProxy will use to connect), and optionally encryption_enabled=True|False.
The EasyRpcServer reference, rpc_server, will be used to register our database functions in a namespace, which will be shared with connecting EasyRpcProxies — more on that in a bit.
db = await Database.create(
database='easy_db',
cache_enabled=True
)Database.create is a factory method which returns a Database() instance, using database=name, cache_enabled=True|False. The default database-type is sqlite, so minimal arguments are required. See aiopyql for more details.
aiopyql is used here for its simplicity, support for asyncio, and caching features which lends itself nicely when justifying the API / Database Layer separation. The patterns used can be translated into any ORM for sharing database access.
# Table Creation
if not 'keystore' in db.tables:
await db.create_table(
'keystore',
[
('key', str, 'UNIQUE NOT NULL'),
('value', str)
],
prim_key='key',
cache_enabled=True
)Our Database() instance ‘db’, automatically discovers table schemas (if any exist) and creates a Table() object within db.tables.
With the above check, a table named ‘keystore’ is created with a very simple key-value schama. This can be thought of persistent python dictionary — with the added bonus of caching.
server.data = {'keystore': db}The Database instance reference is stored in our FastAPI Router — needed for cleanup steps.
# register each table method in table namespace
for table in db.tables:
for func in {'select', 'update', 'insert', 'delete'}:
rpc_server.origin(
getattr(db.tables[table], func), namespace=table)
server.db_server = db_serverEach Table() instance within db.tables contain select, update, insert, and delete methods.
For every function, the EasyRpcServer() instance .origin method is called, using the function reference, and namespace=table_name.
If there is more than one table, the above example would register each of these methods in a namespace matching the table name.
What does register actually do?
The register method “dictionarifies” the functions meta-data, and stores this in a namespace. By meta-data, I mean everything used to call the method(arguments, argument types, method name).
EasyRpcProxy instances use this function meta-data to “mirror” a proxy function, and verify correct input.
@server.on_event('shutdown')
async def shutdown():
await server.data['keystore'].close()The housekeeping shutdown method that allows our instance to gracefully cleanup.
API Layer
from fastapi import FastAPI
from easyrpc.proxy import EasyRpcProxyJust like our Database Layer, the FastAPI router will be used but for Client API requests. EasyRpcProxy for accessing our Database Layer.
@server.on_event('startup')
async def setup():
server.data = {}
server.data['keystore'] = await EasyRpcProxy.create(
'0.0.0.0',
8220,
'/ws/database',
server_secret='abcd1234',
namespace='keystore'
)Within the startup on_event, create a data dictionary within the FastAPI router instance, then create an EasyRpcProxy() instance using host-address, port, using target websocket path, secret and namespace which corresponds with Database Layer EasyRpcServer.
At startup, the EasyRpcProxy() instance will contact the target EasyRpcServer and create proxies of all registered functions in the specified namespace.
aiopyql table methods accept input which is entirely JSON serializable, thus making the functions a perfect fit with EasyRpcProxy -> EasyRpcServer function call constraints.
Insertions & Updates
@server.post("/{table}")
async def insert_or_update_table(table, data: dict):
for key, value in data.items():
exists = not (await server.data['keystore']['select']('*', where={'key': key}) == [] )
if exists:
await server.data['keystore']['update'](
value=value,
where={'key': key}
)
else:
await server.data['keystore']['insert'](
key=key,
value=value
)@server.post(“/{table}”) accepts text input for the table name, and a dictionary containing our Insertions or Updates.
The existence of the primary key is checked before deciding if this is an INSERT / UPDATE operation.
The INSERT / UPDATE methods will behave just as though they were local, syntactically verifying argument input
Selections
@server.get("/{table}")
async def get_table_items(table: str):
return await server.data['keystore']['select']('*')@server.get("/{table}/{key}")
async def get_table_item(table: str, key: str):
return await server.data['keystore']['select'](
'*',
where={'key': key}
)
Here are two implementations of accessing stored data. get_table_items issues a select(*) returning all key values. get_table_item issues a select(*) with a where clause
Deletion
@server.delete("/{table}")
async def delete_table_item(table: str, where: dict):
return await server.data['keystore']['delete'](where=where)This methods accepts both a provided table name and where clause {‘col’: ‘value’}, capable of deleting 1 or many matching table rows.
Starting the Instances
First start the database layer instance
(db-api)$ uvicorn --host 0.0.0.0 --port 8220 db_api_database:server
Then start the API layer instance with optional workers argument:
(db-api)$ uvicorn --host 0.0.0.0 --port 8221 db_api:server --workers 5
To test it out, open http://0.0.0.0:8221/docs/ in a browser.
Each uvicorn worker created corresponds to one isolated version of the API layer code.
uvicorn workers are not able to directly communicate with each other, but in this architecture, share the same database connection, functions, and cache.
When one worker, via a client REST request, makes a change or accesses data, cache is loaded( if enabled), and all uvicorn workers will benefit.
Without this layered separation, each uvicorn worker could create its own connection to a database, but should not use caching if the same database is accessed by another worker.
Many databases can support multiple connections, but few caching layers can ever be trusted if processes cannot communicate changes.
With FastAPI, Developers can quickly and effectively create APIs that are also easy to consume.
Using aiopyql, access to Databases is trivial, and with caching and query queuing, fast.
EasyRpc allows easy extension/sharing of functions and can efficiently allow multiple spawns of an application to share state, connections, and more.
