from responses import HTMLResponsefrom cloudflare import KVStore

WorkerAPI

WorkerAPI is a minimalistic implementation of the FastAPI framework designed to run on Cloudflare Workers. Since Cloudflare Workers now support Python but only with the standard library, this framework provides a way to build simple web service applications that are API-compatible with FastAPI. This allows for easier future migration to FastAPI or even away from Cloudflare Workers if needed.

Motivation

Unlike FastAPI, which launches a server, Cloudflare Workers expose a function that gets called with a request and should return a response. WorkerAPI adapts to this request flow model while keeping the FastAPI-style route definition.

Repository

GitHub: workerapi

Installation

Since WorkerAPI is a minimal framework designed for Cloudflare Workers, it cannot be installed via pip. Instead, you need to add it as a Git submodule:

cd src
git submodule add https://github.com/Deltachaos/cloudflare-python-workerapi workerapi

Then, you can import it in your project as follows:

from workerapi import FastAPI

Example Usage

A simple example using WorkerAPI, that uses KV store, and does HTTP requests:

`src/main.py`

from workerapi import FastAPI, Request
from workerapi.di import Container
from httpx import AsyncClient
from workerapi.cloudflare import KVStore
from workerapi.responses import RedirectResponse, JSONResponse

class MyKVStore(KVStore):
    async def update_something(self, key, data):
        return await self.kv.merge_json(key, {
            "some": data
        })
    
class MyService:
    def __init__(self, store: MyKVStore):
        self.store = store
        
    async def do_something(self, user, body):
        return await self.store.update_something(user, body)
        

c = Container()
c.register(MyKVStore, lambda _: MyKVStore(_.resolve("env").KV_SOME))
c.register(MyService, lambda _: MyService(_.resolve(MyKVStore)))

app = FastAPI(c)

@app.post("/{user}")
async def root(user, request: Request, service: MyService):
    body = await request.body()
    data = await service.do_something(user, body)
    query = request.query
    path = request.path
    content_type = request.headers["content-type"]
    return {"some": "data" + user, "data": data, "query": query, "path": path, "type": content_type}

@app.get("/test")
async def root():
    async with AsyncClient() as client:
        result = await client.post("https://www.example.com", data={
            "some": "form/urlencoded data"
        })
        if result.status_code != 200:
            data = await result.json()
            return JSONResponse({
                "some": data
            }, status_code=500)
    
    return RedirectResponse("https://www.example.com")

on_fetch = app.on_fetch()

`wrangler.toml`

name = "some-app"
main = "src/main.py"
compatibility_flags = ["python_workers"]
compatibility_date = "2024-03-29"

kv_namespaces = [
  { binding = "KV_SOME", id = "<your id>" }
]

Cloudflare Worker Python Support

Cloudflare has recently added support for Python in Workers. You can find more information in the official documentation:

Contributions

Contributions are welcome! Feel free to open an issue or a pull request on the GitHub repository. If you have ideas for improvements or bug fixes, we’d love to hear about them.

License

This project is licensed under the MIT License.