semvec/pypi-packages huggingface dataset:
some-aiohttp-service 0.1.7
Runner framework for aiohttp
License: MIT
Requires Python: <3.13,>=3.9

![Static Badge](https://img.shields.io/badge/Python-3.8%2B-brightgreen?logo=python)
![PyPI - Version](https://img.shields.io/pypi/v/some-aiohttp-service?logo=pypi&logoColor=yellow&color=brightgreen)
![PyPI - License](https://img.shields.io/pypi/l/some-aiohttp-service)
[![codecov](https://codecov.io/gh/tommmlij/some-aiohttp-service/graph/badge.svg?token=GDLHRPUVCG)](https://codecov.io/gh/tommmlij/some-aiohttp-service)
![Static Badge](https://img.shields.io/badge/mypy-checked-blue)
![Static Badge](https://img.shields.io/badge/flake8-linted-green)

Run jobs in the background of an aiohttp server.

## Usage

Define a service

```python
import asyncio
from some_aiohttp_service import BaseService

async def some_long_calculation(a, b):
    await asyncio.sleep(5)
    return f"Done with {a}/{b}"

class TestService(BaseService):
    name = "test"

    @staticmethod
    async def work(job):
        return await some_long_calculation(**job.data)

    async def error_handler(self, job, error):
        print("Error", error)

    async def result_handler(self, job, result):
        print(result)
```

Set up an aiohttp server:

```python
from aiohttp.web import Application, get, run_app
from aiohttp.web_exceptions import HTTPOk, HTTPAccepted

async def health(_):
    raise HTTPOk

async def hello(request):
    a = request.match_info["a"]
    b = request.match_info["b"]
    await request.app["test"].commit_work({"a": a, "b": b})
    raise HTTPAccepted

app = Application()
app.add_routes([get("/work/{a}/{b}", hello), get("/health", health)])

app.cleanup_ctx.append(TestService(app).init)

run_app(app)
```