Awesome

Ketchup

Ketchup is a simple web app for learning how to build web apps using recent Python technologies with an emphasis on asynchronous I/O (asyncio) programming.

Project Source : https://github.com/rockyburt/Ketchup

Part 1 and 2 Source: : https://github.com/rockyburt/Ketchup/tree/part/1-2

Part 3 Source: : https://github.com/rockyburt/Ketchup/tree/part/3

Prerequisites

A Linux or Linux-like work environment (all steps were tested and run on Ubuntu 20.04 inside the WSL environment of Windows 10)
A working Python 3.9 installation

NOTE: Theoretically these steps will run (using platform/OS-equivalent steps) on most modern OS's but the developer's mileage may vary.

`Quart` + `Strawberry-GraphQL` Tutorial

The following tutorial explains how to use Quart and Strawberry-GraphQL to build a simple web application with a GraphQL-based API. All example code is written to heavily lean on asyncio based programming with a focus on including type hint support in a Python setting.

Requirements: Python 3.9 or higher

Part 1: Getting Started with Quart

Quart is a web framework heavily inspired by the Flask framework. It strives to be API-compatible with Flask with method and class signatures only differing to allow for the Python async/await approach to asynchronous I/O programming.

Install Poetry via https://python-poetry.org/docs/master/#installation
Create new Ketchup project
```
poetry new Ketchup
```
Go into the newly created Ketchup directory and add Quart to Poetry's requirements files. Also include Hypercorn for it's ASGI running expertise.
```
cd Ketchup
poetry add Quart hypercorn
```
At the time of writing this doc, the versions installed were:
- Quart 0.15.1

Setup skeleton web app:

Create new file Ketchup/ketchup/webapp.py with the following content:

from quart import Quart

app = Quart(__name__)

@app.route('/')
async def index():
    return 'Hello World'


if __name__ == "__main__":
    app.run()

Test new application by running the following from inside the Ketchup project directory.
```
poetry run python ketchup/webapp.py
```
And open up the following in your browser: http://localhost:5000

Part 2: Adding Strawberry

Strawberry is a library for building GraphQL web applications on top of Python and (preferably but not limited to) asyncio web frameworks. The goal is to define GraphQL schema's using the dataclasses package which is part of the Python stdlib.

Add Strawberry-GraphQL to Poetry's requirements files.
```
poetry add Strawberry-graphql
```
At the time of writing this doc, the versions installed were:
- Strawberry-graphql 0.77.10

Create a new module at Ketchup/ketchup/gqlschema.py

import strawberry


@strawberry.type
class Query:
    @strawberry.field
    def upper(self, val: str) -> str:
        return val.upper()

Create new module for embedding the strawberry graphql endpoint as a standard Quart view as Ketchup/ketchup/strawview.py

import json
import logging
import pathlib
import traceback
from typing import Any, Union

import strawberry
from quart import Response, abort, render_template_string, request
from quart.typing import ResponseReturnValue
from quart.views import View
from strawberry.exceptions import MissingQueryError
from strawberry.file_uploads.utils import replace_placeholders_with_files
from strawberry.http import (GraphQLHTTPResponse, parse_request_data,
                            process_result)
from strawberry.schema import BaseSchema
from strawberry.types import ExecutionResult

logger = logging.getLogger("ketchup")


def render_graphiql_page() -> str:
    dir_path = pathlib.Path(strawberry.__file__).absolute().parent
    graphiql_html_file = f"{dir_path}/static/graphiql.html"

    html_string = None

    with open(graphiql_html_file, "r") as f:
        html_string = f.read()

    return html_string.replace("{{ SUBSCRIPTION_ENABLED }}", "false")


class GraphQLView(View):
    methods = ["GET", "POST"]

    def __init__(self, schema: BaseSchema, graphiql: bool = True):
        self.schema = schema
        self.graphiql = graphiql

    async def process_result(self, result: ExecutionResult) -> GraphQLHTTPResponse:
        if result.errors:
            for error in result.errors:
                err = getattr(error, "original_error", None) or error
                formatted = "".join(traceback.format_exception(err.__class__, err, err.__traceback__))
                logger.error(formatted)

        return process_result(result)

    async def dispatch_request(self, *args: Any, **kwargs: Any) -> Union[ResponseReturnValue, str]:
        if "text/html" in request.headers.get("Accept", ""):
            if not self.graphiql:
                abort(404)

            template = render_graphiql_page()
            return await render_template_string(template)

        content_type = str(request.headers.get("content-type", ""))
        if content_type.startswith("multipart/form-data"):
            form = await request.form
            operations = json.loads(form.get("operations", "{}"))
            files_map = json.loads(form.get("map", "{}"))

            data = replace_placeholders_with_files(operations, files_map, await request.files)
        else:
            data = await request.get_json()

        try:
            request_data = parse_request_data(data)
        except MissingQueryError:
            return Response("No valid query was provided for the request", 400)

        context = {"request": request}

        result = await self.schema.execute(
            request_data.query,
            variable_values=request_data.variables,
            context_value=context,
            operation_name=request_data.operation_name,
            root_value=None,
        )

        response_data = await self.process_result(result)

        return Response(
            json.dumps(response_data),
            status=200,
            content_type="application/json",
        )

Modify file Ketchup/ketchup/webapp.py to have the following content:

import asyncio

from hypercorn.asyncio import serve
from hypercorn.config import Config
from quart import Quart
from strawberry import Schema

from ketchup.gqlschema import Query
from ketchup.strawview import GraphQLView

app = Quart("ketchup")
schema = Schema(Query)


app.add_url_rule("/graphql", view_func=GraphQLView.as_view("graphql_view", schema=schema))


@app.route("/")
async def index():
    return 'Welcome to Ketchup!  Please see <a href="/graphql">Graph<em>i</em>QL</a> to interact with the GraphQL endpoint.'


def hypercorn_serve():
    config = Config()
    config.bind = ["0.0.0.0:5000"]
    config.use_reloader = True
    asyncio.run(serve(app, config, shutdown_trigger=lambda: asyncio.Future()))


if __name__ == "__main__":
    hypercorn_serve()

Test new application by running the following from inside the Ketchup project directory.

a. Run the updated web app
```
poetry run python -m ketchup.webapp
```
b. Open up the following in your browser: http://localhost:5000/graphql

c. Input the following graph query into the left side text area and hit the play button.
```
query {
  upper(val: "dude, where's my car?")
}
```
The result should be (on the right side):
```
{
"data": {
    "upper": "DUDE, WHERE'S MY CAR?"
}    
```

Bonus Points - Running Tests

Add pytest as a dependency.

# The ^6.2 version identifier for pytest is required due to other dependencies pulling down older versions of pytest
poetry add -D "pytest^6.2" pytest-asyncio

Ensure the Ketchup/tests/test_ketchup.py file exists with the following content:

import pytest

from ketchup import __version__, webapp


def test_version():
    assert __version__ == "0.1.0"


@pytest.mark.asyncio
class TestViews:
    async def test_index(self):
        assert "Welcome" in (await webapp.index())

Run the tests by issuing the following from inside the Ketchup directory.

poetry run pytest

The result should be something like:

=============================== test session starts ===============================
platform linux -- Python 3.9.5, pytest-5.4.3, py-1.10.0, pluggy-0.13.1
rootdir: /home/ubuntu/dev/Ketchup
plugins: anyio-3.3.1, asyncio-0.15.1
collected 2 items

tests/test_ketchup.py ..                                                    [100%]

================================ 2 passed in 0.16s ================================

Coding Conventions

It is the author's advice to add the following to help with formatting all code in a standard way.

Add some developer dependencies:

poetry add -D black isort

The standard Python method for activating these formatters would be to append something like the following to Ketchup/pyproject.toml.

exclude = '''
/(
    \.git
  | \.tox
  | \.venv
  | build
  | dist
)/   
'''
line-length = 119  # standard editor width used by github

[tool.isort]
profile = "black"

Part 3a: Persistence with SQLAlchemy and PostgreSQL

This is where the project actually starts getting useful. We are building a ToDo application that can persist todo records to a PostgreSQL database. The following steps assume you are within the Ketchup directory.

Add SQLAlchemy, alembic, and asyncpg as a dependencies.

As of the writing of this tutorial, SQLAlchemy 1.4.23 is the most recent ... with SQLAlchemy 1.4.x the first version to include asyncio support. In addition, asyncio support comes only from PostgreSQL and the asyncpg db adapter.
```
poetry add sqlalchemy alembic asyncpg

# for anyone using mypy/pylance/pyright/vscode ... adding the following should provide better type hint support
# but is optional and won't affect the application
poetry add -D sqlalchemy2.stubs
```

Initialize the Alembic database migration framework.

poetry run alembic init --template async alembic

Modify Ketchup/alembic/env.py so that it uses the same postgres db configuration as the rest of the web app. The file should look like this:

import asyncio
from logging.config import fileConfig

from sqlalchemy import pool
from sqlalchemy.ext.asyncio import create_async_engine

from alembic import context
from ketchup import sqlamodels, base

# this is the Alembic Config object, which provides
# access to the values within the .ini file in use.
config = context.config

# Interpret the config file for Python logging.
# This line sets up loggers basically.
assert config.config_file_name is not None
fileConfig(config.config_file_name)

# add your model's MetaData object here
# for 'autogenerate' support
# from myapp import mymodel
# target_metadata = mymodel.Base.metadata
target_metadata = sqlamodels.mapper_registry.metadata

# other values from the config, defined by the needs of env.py,
# can be acquired:
# my_important_option = config.get_main_option("my_important_option")
# ... etc.


def run_migrations_offline():
    """Run migrations in 'offline' mode.

    This configures the context with just a URL
    and not an Engine, though an Engine is acceptable
    here as well.  By skipping the Engine creation
    we don't even need a DBAPI to be available.

    Calls to context.execute() here emit the given string to the
    script output.

    """
    url = config.get_main_option("sqlalchemy.url")
    context.configure(
        url=url,
        target_metadata=target_metadata,
        literal_binds=True,
        dialect_opts={"paramstyle": "named"},
    )

    with context.begin_transaction():
        context.run_migrations()


def do_run_migrations(connection):
    context.configure(connection=connection, target_metadata=target_metadata)

    with context.begin_transaction():
        context.run_migrations()


async def run_migrations_online():
    """Run migrations in 'online' mode.

    In this scenario we need to create an Engine
    and associate a connection with the context.

    """

    connectable = create_async_engine(
        base.config.DB_URI,
        future=True,
        poolclass=pool.NullPool,
    )

    async with connectable.connect() as connection:
        await connection.run_sync(do_run_migrations)


if context.is_offline_mode():
    run_migrations_offline()
else:
    asyncio.run(run_migrations_online())

Setup the initial SQLAlchemy models by creating Ketchup/ketchup/sqlamodels.py.

import asyncio
import dataclasses
import datetime
import typing
from contextlib import asynccontextmanager

from sqlalchemy import Column, DateTime, Integer, Table, Text
from sqlalchemy.ext.asyncio import (
    AsyncSession,
    async_scoped_session,
    create_async_engine,
)
from sqlalchemy.orm import registry, sessionmaker

from ketchup import base

mapper_registry = registry()

_async_engine = create_async_engine(base.config.DB_URI, future=True)
_async_session_factory = sessionmaker(_async_engine, class_=AsyncSession, expire_on_commit=False)  # type: ignore - having some pyright issues

_get_session = async_scoped_session(_async_session_factory, scopefunc=asyncio.current_task)


@asynccontextmanager
async def atomic_session() -> typing.AsyncIterator[AsyncSession]:
    async with _get_session() as session:
        try:
            yield session
            await session.commit()
        except Exception:
            await session.rollback()
            raise


@mapper_registry.mapped
@dataclasses.dataclass
class Todo:
    __table__ = Table(
        "ketchup_todo",
        mapper_registry.metadata,
        Column(
            "id",
            Integer,
            primary_key=True,
            autoincrement=True,
            nullable=False,
        ),
        Column("text", Text(), nullable=False),
        Column("created", DateTime(True), nullable=False),
        Column("completed", DateTime(True), nullable=True),
    )

    id: int = dataclasses.field(init=False)
    text: str
    created: datetime.datetime = dataclasses.field(
        default_factory=lambda: datetime.datetime.now(datetime.timezone.utc)
    )
    completed: typing.Optional[datetime.datetime] = None

Create new Ketchup/ketchup/base.py file with the following contents for configuration.

import os
import typing


class Config:
    DB_URI: str = "postgresql+asyncpg://localhost/ketchup"

    def __init__(self, prefix: str = "KETCHUP_"):
        for name, type_ in typing.get_type_hints(self).items():
            envname = prefix + name
            if envname in os.environ:
                setattr(self, name, type_(os.environ[envname]))


config = Config()

At this point, make sure PostgreSQL has beeng configured properly with an empty database setup and referenced by either config.DB_URI or by setting os env variable KETCHUP_DB_URI. Once that's been done, use Alembic to auto-generate the first revision migration file.
```
poetry run alembic revision --autogenerate -m "New ketchup_todos table"
```
Run the following to setup the necessary database schema.
```
poetry run alembic upgrade head
```

Part 3b: Making the GraphQL schema useful

Modify Ketchup/ketchup/gqlschema.py to have basic CRUD access via Query and Mutation.

import datetime
import typing

import strawberry
from sqlalchemy.future import select

from ketchup import sqlamodels

strawberry.type(sqlamodels.Todo)


@strawberry.type
class Query:
    todos: list[sqlamodels.Todo]

    @strawberry.field(name="todos")
    async def _todos_resolver(self) -> list[sqlamodels.Todo]:
        async with sqlamodels.atomic_session() as session:
            items = (await session.execute(select(sqlamodels.Todo))).scalars()
        todos = typing.cast(list[sqlamodels.Todo], items)
        return todos


@strawberry.type(description="Standard CRUD operations for todo's")
class TodoOps:
    @strawberry.mutation
    async def add_todo(self, text: str) -> sqlamodels.Todo:
        todo = sqlamodels.Todo(text=text)
        async with sqlamodels.atomic_session() as session:
            session.add(todo)
        return todo

    @strawberry.mutation
    async def remove_todo(self, id: int) -> bool:
        async with sqlamodels.atomic_session() as session:
            item = (await session.execute(select(sqlamodels.Todo).where(sqlamodels.Todo.id == id))).scalars().first()
            todo = typing.cast(typing.Optional[sqlamodels.Todo], item)

            if todo is None:
                return False

            await session.delete(todo)
        return True

    @strawberry.mutation
    async def set_todo_completed(self, id: int, flag: bool = True) -> typing.Optional[sqlamodels.Todo]:
        async with sqlamodels.atomic_session() as session:
            item = (await session.execute(select(sqlamodels.Todo).where(sqlamodels.Todo.id == id))).scalars().first()
            todo = typing.cast(typing.Optional[sqlamodels.Todo], item)

            if todo is None:
                return None

            todo.completed = datetime.datetime.now(datetime.timezone.utc) if flag else None
        return todo

    @strawberry.mutation
    async def modify_todo_text(self, id: int, text: str) -> typing.Optional[sqlamodels.Todo]:
        async with sqlamodels.atomic_session() as session:
            item = (await session.execute(select(sqlamodels.Todo).where(sqlamodels.Todo.id == id))).scalars().first()
            todo = typing.cast(typing.Optional[sqlamodels.Todo], item)

            if todo is None:
                return None

            todo.text = text
        return todo


@strawberry.type
class Mutation:
    todos: TodoOps

    @strawberry.field(name="todos")
    def _todos_resolver(self) -> TodoOps:
        return TodoOps()


schema = strawberry.Schema(query=Query, mutation=Mutation)

The ``Ketchup/ketchup/webapp.py` file will need to be updated to accomodate the SQLAlchemy database access as well as the new mutation support.

import asyncio

from hypercorn.asyncio import serve
from hypercorn.config import Config as HypercornConfig
from quart import Quart

from ketchup import base
from ketchup.gqlschema import schema
from ketchup.strawview import GraphQLView

app = Quart("ketchup")
app.config.from_object(base.config)


app.add_url_rule("/graphql", view_func=GraphQLView.as_view("graphql_view", schema=schema))


@app.route("/")
async def index():
    return 'Welcome to Ketchup!  Please see <a href="/graphql">Graph<em>i</em>QL</a> to interact with the GraphQL endpoint.'


def hypercorn_serve():
    hypercorn_config = HypercornConfig()
    hypercorn_config.bind = ["0.0.0.0:5000"]
    hypercorn_config.use_reloader = True
    asyncio.run(serve(app, hypercorn_config, shutdown_trigger=lambda: asyncio.Future()))


if __name__ == "__main__":
    hypercorn_serve()

At this point it should be possible to restart the web application and start playing with the actual graphql queries.

poetry run python -m ketchup.webapp

And the go to http://localhost:5000/graphql to test queries/mutations. Some examples are:

# Query #1
# The following will show all todos persisted to the database .. upon first query it should return an empty
# result set
query {
  todos {
    id
    text
    created
    completed
  }
}

# Mutation #1
# This will create our first todo record and show ups the generated ID.  After running this at least once
# it should be possible to re-run "Query #1" above and see data that was created and saved.
mutation {
  todos {
    addTodo(text: "Hello World") {
      id
    }
  }
}

Bonus Points - Adding Query Tests

Add new Ketchup/tests/conftest.py file to setup pytest fixtures.

import asyncio
import uuid
from urllib.parse import urlparse, urlunparse

import pytest
from sqlalchemy import pool, text
from sqlalchemy.ext.asyncio import (
    AsyncSession,
    async_scoped_session,
    create_async_engine,
)
from sqlalchemy.orm import close_all_sessions, sessionmaker

from ketchup import __version__, base, sqlamodels


def test_version():
    assert __version__ == "0.1.0"


@pytest.fixture(scope="session")
def event_loop():
    loop = asyncio.get_event_loop_policy().new_event_loop()
    yield loop
    loop.close()


@pytest.fixture(scope="session")
def session_monkeypatch():
    mpatch = pytest.MonkeyPatch()
    yield mpatch
    mpatch.undo()


@pytest.fixture(scope="session")
async def empty_db(session_monkeypatch: pytest.MonkeyPatch):
    parsed = urlparse(base.config.DB_URI)
    dbname = parsed.path[1:] + "_test_" + uuid.uuid4().hex

    newuri = urlunparse([parsed[0], parsed[1], "/template1", parsed[3], parsed[4], parsed[5]])

    anony_engine = create_async_engine(
        newuri,
        future=True,
        poolclass=pool.NullPool,
        isolation_level="AUTOCOMMIT",
    )

    async with anony_engine.connect() as conn:
        await (await conn.execution_options(isolation_level="AUTOCOMMIT")).execute(text(f"CREATE DATABASE {dbname}"))

    async_engine = None
    async_session_factory = None
    make_session = None
    try:
        newuri = urlunparse([parsed[0], parsed[1], "/" + dbname, parsed[3], parsed[4], parsed[5]])
        async_engine = create_async_engine(newuri, future=True)
        async_session_factory = sessionmaker(async_engine, class_=AsyncSession, expire_on_commit=False)  # type: ignore - having some pyright issues
        make_session = async_scoped_session(async_session_factory, scopefunc=asyncio.current_task)
        session_monkeypatch.setattr(sqlamodels, "make_session", make_session)

        async with async_engine.begin() as conn:
            await conn.run_sync(sqlamodels.mapper_registry.metadata.create_all)

        yield
    finally:
        close_all_sessions()
        if async_engine is not None:
            try:
                await async_engine.dispose()
            except Exception:
                ...
        async with anony_engine.connect() as conn:
            await (await conn.execution_options(isolation_level="AUTOCOMMIT")).execute(text(f"DROP DATABASE {dbname}"))

Update the Ketchup/tests/test_ketchup.py file to have the following content.

import pytest

from ketchup import __version__, gqlschema, webapp


def test_version():
    assert __version__ == "0.1.0"


@pytest.mark.asyncio
class TestViews:
    async def test_index(self):
        assert "Welcome" in (await webapp.index())


@pytest.mark.asyncio
@pytest.mark.usefixtures("empty_db")
class TestGraphQuery:
    async def _create(self, text: str) -> int:
        result = await gqlschema.schema.execute('mutation { todos { addTodo(text: "hello world") { id } } }')
        assert result.data is not None
        assert "id" in result.data["todos"]["addTodo"]

        newid = result.data["todos"]["addTodo"]["id"]
        return newid

    async def test_create_remove(self):
        newid = await self._create("hello world")
        result = await gqlschema.schema.execute("mutation { todos { removeTodo(id: %s) } }" % newid)
        assert result.data is not None
        assert result.data["todos"]["removeTodo"] == True

Run the tests by issuing the following from inside the Ketchup directory.

poetry run pytest

The result should be something like:

=============================== test session starts ===============================
platform linux -- Python 3.9.5, pytest-5.4.3, py-1.10.0, pluggy-0.13.1
rootdir: /home/ubuntu/dev/Ketchup
plugins: anyio-3.3.1, asyncio-0.15.1
collected 3 items

tests/test_ketchup.py ...                                                   [100%]

================================ 3 passed in 0.36s ================================

Frameworks/Components Reference

Quart : An asynchronous I/O (asyncio) based web framework inspired by Flask

Strawberry GraphQL : A GraphQL library for Python enabling the building of GraphQL web apps using Python dataclasses

SQLAlchemy : Python ORM mapper (with asyncio support as of v1.4).

asyncpg : A PostgreSQL client library for Python using asyncio.

Alembic : A database migration tool for SQLAlchemy.