Usually, I use Django for building apps in Python, but weeks ago, scrolling on Reddit I found out about Blacksheep.
According to its documentation:
BlackSheep is an asynchronous web framework to build event-based web applications with Python. It is inspired by Flask, ASP.NET Core, and the work by Yury Selivanov.
BlackSheep offers...
A rich code API, based on dependency injection and inspired by Flask and ASP.NET Core A typing-friendly codebase, which enables a comfortable development experience thanks to hints when coding with IDEs Built-in generation of OpenAPI Documentation, supporting version 3, YAML and JSON A cross-platform framework, using the most modern versions of Python Good performance, as demonstrated by the results from TechEmpower benchmarks
I have never used Flask, just Django. But BlackSheep documentation was really easy to follow, and I give it a try. It was easy for me to build an API with it.
The next step was using a database, and I found about Piccolo ORM in BlackSheep's documentation.
According to its Github page:
Piccolo is a fast, user-friendly ORM and query builder which supports asyncio.
Features:
- Support for sync and async.
- A built-in playground, which makes learning a breeze.
- Tab completion support - works great with iPython and VSCode.
- Batteries included - a User model, authentication, migrations, an admin GUI, and more.
- Modern Python - fully type annotated.
- Make your codebase modular and scalable with Piccolo apps (similar to Django apps).
We will build a basic CRUD API using these tools, but first, we set our environment.
mkdir api_tutorial
cd api_tutorial
py -m venv ./myvenv
cd myenv/Scripts
activate
Then we return to api_tutorial folder.
Now we install Piccolo ORM:
pip install "piccolo[postgres]"
Then we create a new piccolo app, in our root folder we run the next command:
piccolo app new sql_app
The last command will generate a new folder in our root folder, with the next structure:
sql_app/
__init__.py
piccolo_app.py
piccolo_migrations/
__init__.py
tables.py
Then, we have to create a new Piccolo project, in our root folder we run:
piccolo project new
After the command above is executed, it will generate the file piccolo_conf.py, with the next code in it:
from piccolo.conf.apps import AppRegistry
from piccolo.engine.postgres import PostgresEngine
DB = PostgresEngine(config={})
# A list of paths to piccolo apps
# e.g. ['blog.piccolo_app']
APP_REGISTRY = AppRegistry(apps=[])
We have to write the configuration of our database in piccolo_conf.py.
piccolo_conf.py
from piccolo.conf.apps import AppRegistry
from piccolo.engine.postgres import PostgresEngine
DB = PostgresEngine(config={
"database": "api_project",
"user": "postgres",
"password": "your password",
"host": "localhost",
"port": 5432,
})
APP_REGISTRY = AppRegistry(apps=['sql_app.piccolo_app'])
Then we create our table in tables.py located in the sql_app folder.
from piccolo.table import Table
from piccolo.columns import Varchar, Integer
class Expense(Table):
amount = Integer()
description = Varchar()
In piccolo_app.py we register our app's table. We will use table_finder to automatically import any Table subclasses from a given list of modules. Here there are more details.
piccolo_app.py
import os
from piccolo.conf.apps import AppConfig, table_finder
CURRENT_DIRECTORY = os.path.dirname(os.path.abspath(__file__))
APP_CONFIG = AppConfig(
app_name="sql_app",
migrations_folder_path=os.path.join(
CURRENT_DIRECTORY, "piccolo_migrations"
),
table_classes=table_finder(modules=["sql_app.tables"], exclude_imported=True),
migration_dependencies=[],
commands=[],
)
We are using Postgres, we need to create our database, so, open your command line and create the database:
psql -U <your user>
create database api_project;
After the database is created, we need to run migrations, in our root folder, run:
piccolo migrations new sql_app --auto
piccolo migrations forwards sql_app
When we run piccolo migrations new sql_app we should see the next message in your command line:
- Creating new migration ...
- Created tables 1
- Dropped tables 0
- Renamed tables 0
- Created table columns 2
- Dropped columns 0
- Columns added to existing tables 0
- Renamed columns 0
- Altered columns 0`
And when we run piccolo migrations forwards sql_app, we should see the next message:
👍 1 migration already complete
⏩ 1 migration not yet run
🚀 Running 1 migration:
- 2022-06-04T15:29:20:540168 [forwards]... ok! ✔️
Also, it should be a new file with a timestamp in the migrations folder.
Now is time to install Blacksheep:
pip install blacksheep uvicorn
As Blacksheep's documentation mentions, "BlackSheep belongs to the category of ASGI web frameworks, therefore it requires an ASGI HTTP server to run, such as uvicorn, or hypercorn". In its tutorial uvicorn is used, so we will use uvicorn too.
In our root folder, we create main.py and write the next code:
if __name__ == "__main__":
import uvicorn
uvicorn.run("app:app", reload=True)
Now, to run our server, we just need to write in our command-line py main.py, instead uvicorn server:app --reload
It is time to write the endpoints. First, we will write three endpoints, two GET endpoints, and one POST endpoint.
We create a new file, app.py, to write our endpoints.
import typing
from piccolo.utils.pydantic import create_pydantic_model
from piccolo.engine import engine_finder
from blacksheep import Application, FromJSON, json, Response, Content
from sql_app.tables import Expense
ExpenseModelIn: typing.Any = create_pydantic_model(table=Expense, model_name=" ExpenseModelIn")
ExpenseModelOut: typing.Any = create_pydantic_model(table=Expense, include_default_columns=True, model_name=" ExpenseModelOut")
ExpenseModelPartial: typing.Any = create_pydantic_model(
table=Expense, model_name="ExpenseModelPartial", all_optional=True
)
app = Application()
@app.router.get("/expenses")
async def expenses():
try:
expense = await Expense.select()
return expense
except:
return Response(404, content=Content(b"text/plain", b"Not Found"))
We create three pydantic models, to validate our data. ExpenseModelIn is the model or structure we use when we add new data to the database when we are using the POST method. If we try to enter data without the fields we specified in our table, it will return an error. For example:
This is correct and will pass.
{"amount":20, "description":"Food"}
Any json without the fields "amount" and "description" will not pass.
ExpenseModelPartial is the model a function received as an argument when we want to modify data in the database, for example, when using the PATCH method.
ExpenseModelOut is the model that a function returns when using any endpoint and includes defaults columns, like id.
We initialize our application by calling the Application class. Then, we use the decorator @app.router.get() and "/expenses" as a route parameter. Next, we define a function that returns all the entries in our database, using the select() method from Piccolo ORM.
GET
@app.router.get("/expense/{id}")
async def expense(id: int):
expense = await Expense.select().where(id==Expense.id)
if not expense:
return Response(404, content=Content(b"text/plain", b"Id not Found"))
return expense
This other endpoint uses the GET method to return a specific entry, selected by its id. We define a function with id as a parameter. Then, we use select() and where(). If the id is not in the database, it will return the status code 404.
POST
@app.router.post("/expense")
async def create_expense(expense_model: FromJSON[ExpenseModelIn]):
try:
expense = Expense(**expense_model.value.dict())
await expense.save()
return ExpenseModelOut(**expense.to_dict())
except:
return Response(400, content=Content(b"text/plain", b"Bad Request"))
We use the decorator @app.router.post() and ("/expense") as parameter. We define a function to create a new entry in the database that has an ExpenseModelIn as a parameter. The function passes the JSON data converted in a dictionary to the Expense constructor as a keyword argument and saves it. And returns the data with the id(ExpenseModelOut). If the request has not had the same fields defined in ExpenseModelIn, it will send a Bad request as a response.
Below our post route, to initialize our database and use the connection pool to make queries.
async def open_database_connection_pool(application):
try:
engine = engine_finder()
await engine.start_connection_pool()
except Exception:
print("Unable to connect to the database")
async def close_database_connection_pool(application):
try:
engine = engine_finder()
await engine.close_connection_pool()
except Exception:
print("Unable to connect to the database")
app.on_start += open_database_connection_pool
app.on_stop += close_database_connection_pool
Now we can run the server:
py main.py
PATCH
@app.router.patch("expense/{id}")
async def patch_expense(
id: int, expense_model: FromJSON[ExpenseModelPartial]
):
expense = await Expense.objects().get(Expense.id == id)
if not expense:
return Response(404, content=Content(b"text/plain", b"Id not Found"))
for key, value in expense_model.value.dict().items():
if value is not None:
setattr(expense, key, value)
await expense.save()
return ExpenseModelOut(**expense.to_dict())
In the patch method, we use the id and ExpenseModelPartial as parameter, it will update the entry with the id we want, save it, and send the JSON with the entry updated. Otherwise, it will send an error and "Id Not Found" as a response.
PUT
@app.router.put("/expense/{id}")
async def put_expense(
id: int, expense_model: FromJSON[ExpenseModelIn]
):
expense = await Expense.objects().get(Expense.id == id)
if not expense:
return Response(404, content=Content(b"text/plain", b"Id Not Found"))
for key, value in expense_model.value.dict().items():
if value is not None:
setattr(expense, key, value)
await expense.save()
return ExpenseModelOut(**expense.to_dict())
In the put method, we use the id and ExpenseModelIn as parameters, it will update the entry with the id we want, save it, and send the JSON with the entry updated. Otherwise, it will send an error and "Id Not Found" as a response.
DELETE
@app.router.delete("/expense/{id}")
async def delete_expense(id: int):
expense = await Expense.objects().get(Expense.id == id)
if not expense:
return Response(404, content=Content(b"text/plain", b"Id Not Found"))
await expense.remove()
return json({"message":"Expense deleted"})
For the DELETE method we pass the id of the entry we want to delete and check if it is in the database, and remove it using the method remove() from Piccolo ORM. Otherwise, it will send an error and "Id Not Found" as a response.
Conclusion
Honestly, it was the first time a tried another web framework in Python besides Django. I will not compare them, I think they are different. Also, it was the first time using Piccolo and Postgres, I have been using SQLite, but I read in an issue in Piccolo's GitHub that connection pool isn't enabled in SQLite. For me, it was fun to learn Blacksheep and Piccolo, and I will start to learn more and build projects with them.
Thank you for taking your time and read this article.
If you have any recommendations about other packages, architectures, how to improve my code, my English, or anything; please leave a comment or contact me through Twitter, LinkedIn.
The source code is here.