Tutorial¶

Let’s dig a little deeper and learn the basic concepts of Zayt.

We will create a greeting api that logs the greet requests.

Installing Zayt¶

Before going any further, we need to install Zayt and uvicorn.

$ pip install zayt uvicorn[standard]

Structure of the application¶

A zayt application can be structured in several ways:

Minimal¶

project/
├── application.py
├── configuration/
│   └── settings.py
└── resources/

With modules¶

project/
├── application/
│   ├── __init__.py
│   ├── handler.py
│   ├── repository.py
│   └── service.py
├── configuration/
│   └── settings.py
└── resources/

And… that’s it! A module or package named application will automatically be imported and scanned for handlers and services.

You can structure the application package however suits you.

Running the application¶

We will use uvicorn to run the application and automatically reload when we make changes to the code:

$ uvicorn zayt.run:app --reload
INFO:     Will watch for changes in these directories: ['/home/user/projects/zayt-tutorial']
INFO:     Uvicorn running on http://127.0.0.1:8000 (Press CTRL+C to quit)
INFO:     Started reloader process [1001] using WatchFiles
INFO:     Started server process [1000]
INFO:     Waiting for application startup.
INFO:     Application startup complete.

Handler functions¶

Handler functions will respond to HTTP or WebSocket requests. They can receive services through dependency injection.

application/handler.py¶

from asgikit import Request
from zayt.web import get

@get("hello/{name}")
async def hello(request: Request):
    name = request.path_params["name"]
    await request.respond_json({"greeting": f"Hello, {name}!"})

@get("hello/{name}") defines the function as a handler on the given path. If no path is given, the root path (“/”) will be used.

{name} defines a path parameter that will be stored in request.path_params, which is a dict object.

And now we test if our handler is working:

$ curl localhost:8000/hello/World
{"greeting": "Hello, World!"}

Right now our handler just get a name from the path and respond with JSON data to the client.

Creating the Greeter service¶

Our service will have a method that receives a name and returns a greeting. It will be injected into the handler we created previously.

application/service.py¶

from zayt.di import service


@service
class Greeter:
    def greet(self, name: str) -> str:
        return f"Hello, {name}!"

@service registers the class in the dependency injection system so it can be injected in other classes or handler functions

application/handler.py¶

from asgikit import Request
from zayt.web import get
from .service import Greeter


@get("/hello/{name}")
async def hello(request: Request, greeter: Greeter):
    name = request.path_params["name"]
    greeting = greeter.greet(name)
    await request.respond_json({"greeting": greeting})

The parameter greeter will be injected into the handler.

Adding a database¶

Our greeting application is working fine, but we might want to register the greeting requests in a persistent database, for auditing purposes.

To do this we need to create the database service and inject it into the Greeter service. For this we can use the aiosqlite:

$ pip install aiosqlite

aiosqlite provides a class called Connection. However, we can not decorate it with @service, so in this case we need to create a factory function for it:

application/repository.py¶

from datetime import datetime
from typing import Annotated
import aiosqlite
from zayt.di import service, Inject

@service # (1)
async def connection_factory() -> aiosqlite.Connection:
    conn = await aiosqlite.connect("sqlite:///db.sqlite3")
    yield conn
    await conn.close()


@service
class GreetingRepository:
    database: Annotated[aiosqlite.Connection, Inject] # (2)

    async def initialize(self): # (3)
        await self.database.execute(
            """
            create table if not exists greeting_log(
                greeting text not null,
                datetime text not null
            );
            """
        )

    async def finalize(self): # (4)
        await self.database.execute("drop table if exists greeting_log;")

    async def save_greeting(self, greeting: str, date: datetime):
        query = """
            insert into greeting_log (greeting, datetime)
            values (:greeting, datetime(:datetime))
        """
        params = {"greeting": greeting, "datetime": date}
        await self.database.execute(query, params)

Note

A function decorated with @service is used to create a service when you need to provide types you do not own
Inject the aiosqlite.Connection service in the GreetingRepository
A method called initialize will be called after the service is constructed in order to run any initialization logic
A method called finalize will be called before the service is destroyed in order to run any cleanup logic

application/handler.py¶

from datetime import datetime
from asgikit import Request
from zayt.web import get
from .repository import GreetingRepository
from .service import Greeter


@get("hello/{name}")
async def hello_name(
    request: Request,
    greeter: Greeter,
    repository: GreetingRepository,
):
    name = request.path_params["name"]
    greeting = greeter.greet(name)
    await repository.save_greeting(greeting, datetime.now())
    await request.respond_json({"greeting": greeting})

Execute actions after response¶

The greetings are being saved to the database, but now we have a problem: the user has to wait until the greeting is saved before receiving it.

To solve this problem and improve the user experience, we can save the greeting after the request is completed:

application/handler.py¶

from datetime import datetime
from asgikit import Request
from zayt.web import get
from .repository import GreetingRepository
from .service import Greeter


@get("hello/{name}")
async def hello_name(
    request: Request,
    greeter: Greeter,
    repository: GreetingRepository,
):
    name = request.path_params["name"]
    greeting = greeter.greet(name)
    await request.respond_json({"greeting": greeting}) # (1)

    await repository.save_greeting(greeting, datetime.now()) # (2)

Note

The call to respond_json completes the response
The greeting is saved after the response is completed

Retrieving the greeting logs¶

To see the greetings saved to the database, we just need to add a handler to get the logs and return them:

application/repository.py¶

@service
class GreetingRepository:
    # ...
    async def get_greetings(self) -> list[dict[str, str]]:
        query = """
            select l.greeting, datetime(l.datetime) from greeting_log l
            order by rowid desc
        """

        async with self.database.execute(query) as cur:
            rows = await cur.fetchall()

        return [
            {"greeting": row[0], "datetime": row[1]}
            for row in rows
        ]

application/handler.py¶

# ...
@get("/logs")
async def greeting_logs(request: Request, repository: GreetingRepository):
    greetings = await repository.get_greetings()
    await request.respond_json(greetings)

Now let us try requesting some greetings and retrieving the logs:

$ curl localhost:8000/hello/Python
{"greeting": "Hello, World!"}

$ curl localhost:8000/hello/World
{"greeting": "Hello, Python!"}

$ curl -s localhost:8000/logs | python -m json.tool
[
    {
        "greeting": "Hello, World!",
        "datetime": "2025-01-01 12:00:10"
    },
    {
        "greeting": "Hello, Python!",
        "datetime": "2025-01-01 12:00:20"
    },
]

Receiving POST data¶

We can also send the name in the body of the request, instead of the url, and use pydantic to parse the request body:

application/handler.py¶

from zayt.web import get, post
# ...
@post("hello")
async def hello_post(
    request: Request,
    greeter: Greeter,
    repository: GreetingRepository,
):
    greeting_request = await request.body.read_json()
    name = greeting_request["name"]
    greeting = greeter.greet(name)
    await request.respond_json({"greeting": greeting})
    await repository.save_greeting(greeting, datetime.now())

And to test it:

$ curl \
    -H 'Content-Type: application/json' \
    -d '{"name": "World"}' \
    localhost:8000/hello
{"greeting": "Hello, World!"}

Tutorial¶

Installing Zayt¶

Structure of the application¶

Running the application¶

Handler functions¶

Creating the Greeter service¶

Adding a database¶

Execute actions after response¶

Retrieving the greeting logs¶

Receiving POST data¶

Zayt

Navigation

Related Topics