DEV Community 👩‍💻👨‍💻

Carlos Armando Marcano Vargas
Carlos Armando Marcano Vargas

Posted on • Updated on

A simple REST API with Blacksheep and Piccolo ORM | Python.

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
Enter fullscreen mode Exit fullscreen mode
py -m venv ./myvenv
Enter fullscreen mode Exit fullscreen mode
cd myenv/Scripts
activate
Enter fullscreen mode Exit fullscreen mode

Then we return to api_tutorial folder.

Now we install Piccolo ORM:

pip install "piccolo[postgres]"
Enter fullscreen mode Exit fullscreen mode

Then we create a new piccolo app, in our root folder we run the next command:

piccolo app new sql_app

Enter fullscreen mode Exit fullscreen mode

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

Enter fullscreen mode Exit fullscreen mode

Then, we have to create a new Piccolo project, in our root folder we run:

piccolo project new

Enter fullscreen mode Exit fullscreen mode

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=[])

Enter fullscreen mode Exit fullscreen mode

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'])

Enter fullscreen mode Exit fullscreen mode

Then we create our table in tables.py located in the sql_app folder.

tables.py

from piccolo.table import Table
from piccolo.columns import Varchar, Integer

class Expense(Table):
    amount = Integer()
    description = Varchar()

Enter fullscreen mode Exit fullscreen mode

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=[],
)

Enter fullscreen mode Exit fullscreen mode

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;

Enter fullscreen mode Exit fullscreen mode

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

Enter fullscreen mode Exit fullscreen mode

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`

Enter fullscreen mode Exit fullscreen mode

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! ✔️

Enter fullscreen mode Exit fullscreen mode

Also, it should be a new file with a timestamp in the migrations folder.

Now is time to install Blacksheep:

pip install blacksheep uvicorn

Enter fullscreen mode Exit fullscreen mode

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:

main.py

if __name__ == "__main__":

    import uvicorn

    uvicorn.run("app:app", reload=True)

Enter fullscreen mode Exit fullscreen mode

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.

app.py

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=" ExpenseModelIn")

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"))

Enter fullscreen mode Exit fullscreen mode

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.

Enter fullscreen mode Exit fullscreen mode

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

Enter fullscreen mode Exit fullscreen mode

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"))

Enter fullscreen mode Exit fullscreen mode

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

Enter fullscreen mode Exit fullscreen mode

Now we can run the server:

py main.py
Enter fullscreen mode Exit fullscreen mode

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())

Enter fullscreen mode Exit fullscreen mode

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())

Enter fullscreen mode Exit fullscreen mode

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"})

Enter fullscreen mode Exit fullscreen mode

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.

References:

Top comments (0)

🌚 Browsing with dark mode makes you a better developer by a factor of exactly 40.

It's a scientific fact.