We already created a shared layout for the web interface of our API and fixed the problem with the static resources. Now we can create a dashboard that shows us some important numbers about our application.
This post is part of my journey to learn Python. You find the code for this post in my PythonFriday repository on GitHub.
Create a Pydantic model for the statistics
We can start our work with a StatisticOverview class and put it into the new file models/statistics.py:
|
1 2 3 4 5 6 |
from pydantic import BaseModel class StatisticOverview(BaseModel): total_tasks: int total_done: int total_open: int |
This model helps us to transfer our numbers from our data store to the web interface.
Create a test for the new behaviour
To fill our StatisticOverview model with data, we will need a new method on our data store. Before we add that, we create a test that should check if the data we get back makes any sense:
|
1 2 3 4 5 6 7 8 9 10 |
def test_fetches_statistics(with_db): store = DataStoreDb(with_db) store.add(TaskInput(name="counter", priority=1, due_date=date.today(), done=False)) stats = store.get_statistics() assert stats.total_tasks == stats.total_open + stats.total_done assert stats.total_tasks >= 1 |
Implement the statistic gathering
With the test in place, we can now implement the get_statistics() method inside the data/datastore_db.py file:
|
1 2 3 4 5 6 7 8 9 10 11 12 13 |
def get_statistics(self) -> StatisticOverview: query = ( select( func.count("*").label("total"), func.count("*").filter(Task.done==True).label("done"), func.count("*").filter(Task.done==False).label("open") , ) ) result = self.db.execute(query).first() return StatisticOverview(total_tasks=result[0], total_done=result[1], total_open=result[2]) |
Unfortunately, there is no simple way to count rows in SQLAlchemy and I had to use a lot of magic to get the numbers we are interested in. The code above runs three times a SELECT COUNT(*), a first time to count all the tasks, a second time for all tasks that are done and a last time for all tasks that are still open. We get the data back from the database and put it into our Pydantic model.
Extract the dependency get_db()
We connect to the database through our get_db() method that is inside our routers/todo.py file. Since we want to put our dashboard endpoint somewhere else, we first need to move the get_db() method in a location we can access from both places. We can create a file dependencies.py next to main.py and move the method to its new location:
|
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 |
import os from .data.database import create_session_factory from .data.datastore_db import DataStoreDb async def get_db(): """ Creates the datastore """ db_file = os.path.join( os.path.dirname(__file__), '.', 'db', 'todo_api.sqlite') factory = create_session_factory(db_file) session = factory() db = DataStoreDb(session) try: yield db finally: session.close() |
We not only need to fix the import in our router, but also in our tests. We can now run the tests and the dependency replacement should still work.
Add a test for the dashboard
Before we implement the dashboard, we create a test to check that our dashboard contains the right data and that the total number matches the sum of open and done tasks:
|
1 2 3 4 5 6 7 8 9 |
def test_dashboard(): response = client.get("/dashboard") assert response.status_code == 200 soup = BeautifulSoup(response.text, 'html.parser') assert soup.title.text == "Dashboard To-Do Task API" assert soup.body.h1.text == "Dashboard" numbers = re.findall(r"\d+", soup.body.p.text) assert int(numbers[0]) == int(numbers[1]) + int(numbers[2]) |
Add the dashboard endpoint
In our main.py file we reference our newly moved get_db() method and write the code for our dashboard endpoint:
|
1 2 3 4 5 6 7 8 |
from .dependencies import get_db @app.get("/dashboard", include_in_schema=False) async def dashboard(request: Request, db: DataStoreDb = Depends(get_db)): stats = db.get_statistics() return templates.TemplateResponse( request=request, name="dashboard.html", context=jsonable_encoder(stats) ) |
The jsonable_encoder() method turns our Pydantic model into a JSON structure that allows us to use the field names inside {{ }} blocks to show the values in our new template/dashboard.html file:
|
1 2 3 4 5 6 7 8 9 10 11 12 |
{% extends "./shared/layout.html" %} {% block title %}Dashboard To-Do Task API{% endblock %} {% block main %} <h1>Dashboard</h1> <p>We have {{total_tasks}} tasks in our system. <ul> <li>Open: {{total_open}}</li> <li>Done: {{total_done}}</li> </ul> </p> {% endblock %} |
With the template in place, we can run our tests, and everything should pass.
Next
With our little dashboard in place, we can share some basic data about our application in a user-friendly way. We can extend on the templates and the static route and create a full API-driven web application.
While writing the query for the statistics method, I noticed that our SQLAlchemy implementation still runs synchronously. Next week I show you all the necessary steps to use an asynchronous SQLAlchemy.
2 thoughts on “Python Friday #238: Create a Dashboard for FastAPI”