woggioni/kaya
1
0
Fork 0
Code Actions Packages Releases Activity

Initial commit

Browse Source
This commit is contained in:
woggioni
2026-07-15 09:18:48 +08:00
commit 3e804607fe
26 changed files with 1779 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files
.gitea/workflows/build.yaml
+39
View File
@@ -0,0 +1,39 @@
name: CI
on:
push:
tags:
- '*'
jobs:
build_pip_package:
name: "Build Pip package"
runs-on: hostinger
timeout-minutes: 30
container:
image: gitea.woggioni.net/woggioni/gitea-runner/ubuntu:26.04
steps:
- name: Checkout sources
uses: actions/checkout@v6
with:
fetch-depth: 0
fetch-tags: true
- uses: actions/setup-python@v6
with:
python-version: '3.14.6'
cache: 'pip'
- name: Install dependencies
run: |
python -m venv .venv
.venv/bin/pip install -r requirements-dev.txt
- name: Unit tests
run: |
.venv/bin/pip install .
.venv/bin/python -m mypy -p src
.venv/bin/python -m unittest discover -s tests
- name: Publish artifacts
env:
TWINE_REPOSITORY_URL: ${{ vars.PYPI_REGISTRY_URL }}
TWINE_USERNAME: ${{ vars.PUBLISHER_USERNAME }}
TWINE_PASSWORD: ${{ secrets.PUBLISHER_TOKEN }}
run: |
.venv/bin/pyproject-build
.venv/bin/twine upload --repository gitea dist/*.whl dist/*.tar.gz
.gitignore
+8
View File
@@ -0,0 +1,8 @@
.venv
__pycache__
*.pyc
.mypy_cache
_version.py
*.egg-info
/build
/dist
LICENSE
+18
View File
@@ -0,0 +1,18 @@
MIT License
Copyright (c) 2026 woggioni
Permission is hereby granted, free of charge, to any person obtaining a copy of this software and
associated documentation files (the "Software"), to deal in the Software without restriction, including
without limitation the rights to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
copies of the Software, and to permit persons to whom the Software is furnished to do so, subject to the
following conditions:
The above copyright notice and this permission notice shall be included in all copies or substantial
portions of the Software.
THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR IMPLIED, INCLUDING BUT NOT
LIMITED TO THE WARRANTIES OF MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO
EVENT SHALL THE AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER
IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE
USE OR OTHER DEALINGS IN THE SOFTWARE.
README.md
+35
View File
@@ -0,0 +1,35 @@
# kaya
A lightweight ASGI/RSGI web framework with method-aware routing, path matching, and recursive wildcard support.
## Build & run locally
Install dev dependencies:
```bash
pip install --index-url https://gitea.woggioni.net/api/packages/woggioni/pypi/simple --extra-index-url https://pypi.org/simple -r requirements-dev.txt
```
Build the package:
```bash
python -m build
```
Run the example:
```bash
python example/hello.py
```
## Tests
```bash
PYTHONPATH=src python -m unittest discover tests
```
## Static analysis
```bash
mypy src/kaya
```
conf/logging.json
+46
View File
@@ -0,0 +1,46 @@
{
"version": 1,
"disable_existing_loggers": false,
"handlers": {
"console": {
"class": "logging.StreamHandler",
"formatter": "default",
"level": "DEBUG",
"stream": "ext://sys.stderr"
},
"access": {
"class": "logging.StreamHandler",
"formatter": "access",
"level": "DEBUG",
"stream": "ext://sys.stdout"
}
},
"formatters": {
"default": {
"format": "{asctime}.{msecs:0<3.0f} [{levelname}] ({processName:s}/{threadName:s}) - {name} - {message}",
"style": "{",
"datefmt": "%Y-%m-%d %H:%M:%S"
},
"access": {
"format": "%(message)s"
}
},
"loggers": {
"root": {
"handlers": [
"console"
]
},
"_granian": {
"level": "DEBUG",
"propagate": false
},
"granian.access": {
"handlers": [
"access"
],
"level": "DEBUG",
"propagate": false
}
}
}
example/hello.py
+45
View File
@@ -0,0 +1,45 @@
from kaya import BugisApp, HttpContext, HttpMethod
from typing import List
class Hello(BugisApp):
pass
# async def handle_request(self, ctx: HttpContext) -> None:
# async for chunk in ctx.request_body:
# print(chunk.decode())
# await ctx.send_str(200, 'Hello World')
app = BugisApp()
@app.GET('/hello')
@app.GET('/hello2')
async def handle_request(ctx: HttpContext) -> None:
async for chunk in ctx.request_body:
print(chunk.decode())
await ctx.send_str(200, 'Hello World')
@app.GET('/hello/alice')
async def hello_alice(ctx: HttpContext) -> None:
async for _ in ctx.request_body:
pass
await ctx.send_str(200, 'Hello Alice 2')
@app.GET('/hello/*')
async def hello_wildcard(ctx: HttpContext, args: List[str]) -> None:
async for _ in ctx.request_body:
pass
await ctx.send_str(200, f'Hello {args[0]}')
@app.GET('/hello/bob/marley')
async def hello_bob_marley(ctx: HttpContext) -> None:
async for _ in ctx.request_body:
pass
await ctx.send_str(200, f'Hello bob Marley')
@app.route('/hello/*', HttpMethod.PUT, recursive=True)
async def handle_put_request(ctx: HttpContext, _: List[str]) -> None:
async for chunk in ctx.request_body:
print(chunk.decode())
await ctx.send_str(200, 'Message received')
pyproject.toml
+53
View File
@@ -0,0 +1,53 @@
[build-system]
requires = ["setuptools>=61.0", "setuptools-scm>=8"]
build-backend = "setuptools.build_meta"
[project]
name = "kaya"
dynamic = ["version"]
authors = [
{ name="Walter Oggioni", email="oggioni.walter@gmail.com" },
]
description = "A lightweight ASGI/RSGI web framework"
readme = "README.md"
requires-python = ">=3.10"
classifiers = [
'Development Status :: 3 - Alpha',
'Topic :: Utilities',
'License :: OSI Approved :: MIT License',
'Intended Audience :: System Administrators',
'Intended Audience :: Developers',
'Environment :: Console',
'License :: OSI Approved :: MIT License',
'Programming Language :: Python :: 3',
]
dependencies = [
"pwo",
]
[project.optional-dependencies]
dev = [
"build", "mypy", "ipdb", "twine", "granian", "httpx"
]
rsgi = [
"granian"
]
[project.urls]
"Homepage" = "https://github.com/woggioni/kaya"
"Bug Tracker" = "https://github.com/woggioni/kaya/issues"
[tool.mypy]
python_version = "3.12"
disallow_untyped_defs = true
show_error_codes = true
no_implicit_optional = true
warn_return_any = true
warn_unused_ignores = true
exclude = ["scripts", "docs", "test"]
strict = true
[tool.setuptools_scm]
version_file = "src/kaya/_version.py"
requirements-dev.txt
+154
View File
@@ -0,0 +1,154 @@
#
# This file is autogenerated by pip-compile with Python 3.14
# by the following command:
#
# pip-compile --allow-unsafe --extra=dev --extra=rsgi --output-file=requirements-dev.txt pyproject.toml
#
--index-url https://gitea.woggioni.net/api/packages/woggioni/pypi/simple
--extra-index-url https://pypi.org/simple
anyio==4.14.1
# via httpx
ast-serialize==0.5.0
# via mypy
asttokens==3.0.1
# via stack-data
build==1.5.0
# via kaya (pyproject.toml)
certifi==2026.6.17
# via
# httpcore
# httpx
# requests
cffi==2.0.0
# via cryptography
charset-normalizer==3.4.7
# via requests
click==8.4.2
# via granian
cryptography==49.0.0
# via secretstorage
decorator==5.3.1
# via
# ipdb
# ipython
docutils==0.23
# via readme-renderer
executing==2.2.1
# via stack-data
granian==2.7.7
# via kaya (pyproject.toml)
h11==0.16.0
# via httpcore
httpcore==1.0.9
# via httpx
httpx==0.28.1
# via kaya (pyproject.toml)
id==1.6.1
# via twine
idna==3.18
# via
# anyio
# httpx
# requests
ipdb==0.13.13
# via kaya (pyproject.toml)
ipython==9.15.0
# via ipdb
ipython-pygments-lexers==1.1.1
# via ipython
jaraco-classes==3.4.0
# via keyring
jaraco-context==6.1.2
# via keyring
jaraco-functools==4.5.0
# via keyring
jedi==0.20.0
# via ipython
jeepney==0.9.0
# via
# keyring
# secretstorage
keyring==25.7.0
# via twine
librt==0.11.0
# via mypy
markdown-it-py==4.2.0
# via rich
matplotlib-inline==0.2.2
# via ipython
mdurl==0.1.2
# via markdown-it-py
more-itertools==11.1.0
# via
# jaraco-classes
# jaraco-functools
mypy==2.1.0
# via kaya (pyproject.toml)
mypy-extensions==1.1.0
# via mypy
nh3==0.3.6
# via readme-renderer
packaging==26.2
# via
# build
# twine
parso==0.8.7
# via jedi
pathspec==1.1.1
# via mypy
pexpect==4.9.0
# via ipython
prompt-toolkit==3.0.52
# via ipython
psutil==7.2.2
# via ipython
ptyprocess==0.7.0
# via pexpect
pure-eval==0.2.3
# via stack-data
pwo==0.1.2
# via kaya (pyproject.toml)
pycparser==3.0
# via cffi
pygments==2.20.0
# via
# ipython
# ipython-pygments-lexers
# readme-renderer
# rich
pyproject-hooks==1.2.0
# via build
readme-renderer==45.0
# via twine
requests==2.34.2
# via
# requests-toolbelt
# twine
requests-toolbelt==1.0.0
# via twine
rfc3986==2.0.0
# via twine
rich==15.0.0
# via twine
secretstorage==3.5.0
# via keyring
stack-data==0.6.3
# via ipython
traitlets==5.15.1
# via
# ipython
# matplotlib-inline
twine==6.2.0
# via kaya (pyproject.toml)
typing-extensions==4.15.0
# via
# mypy
# pwo
urllib3==2.7.0
# via
# id
# requests
# twine
wcwidth==0.8.1
# via prompt-toolkit
requirements.txt
+17
View File
@@ -0,0 +1,17 @@
#
# This file is autogenerated by pip-compile with Python 3.14
# by the following command:
#
# pip-compile --allow-unsafe --extra=rsgi --output-file=requirements.txt pyproject.toml
#
--index-url https://gitea.woggioni.net/api/packages/woggioni/pypi/simple
--extra-index-url https://pypi.org/simple
click==8.4.2
# via granian
granian==2.7.7
# via kaya (pyproject.toml)
pwo==0.1.2
# via kaya (pyproject.toml)
typing-extensions==4.15.0
# via pwo
src/kaya/__init__.py
+15
View File
@@ -0,0 +1,15 @@
from ._app import BugisApp
from ._http_method import HttpMethod
from ._http_context import HttpContext
from ._tree import Tree, PathIterator
from ._path_handler import PathHandler
__all__ = [
'HttpMethod',
'BugisApp',
'HttpContext',
'Tree',
'PathHandler',
'PathIterator'
]
src/kaya/_app.py
+149
View File
@@ -0,0 +1,149 @@
from abc import ABC, abstractmethod
from asyncio import Queue, AbstractEventLoop
from asyncio import get_running_loop
from logging import getLogger
from typing import Callable, Awaitable, Any, Mapping, Sequence, Optional, Unpack, Tuple, TYPE_CHECKING
from pathlib import Path, PurePath
from pwo import Maybe, AsyncQueueIterator
from hashlib import md5
from ._http_context import HttpContext
from ._http_method import HttpMethod
from ._types import StrOrStrings
from base64 import b64encode, b64decode
from mimetypes import guess_type
if TYPE_CHECKING:
from _typeshed import StrOrBytesPath
try:
from ._rsgi import RsgiContext
from granian._granian import RSGIHTTPProtocol, RSGIHTTPScope # type: ignore
except ImportError:
pass
from ._asgi import AsgiContext
from ._tree import Tree
from ._types.asgi import LifespanScope, HTTPScope as ASGIHTTPScope, WebSocketScope
log = getLogger(__name__)
type HttpHandler = Callable[[HttpContext, Unpack[Any]], Awaitable[None]]
class AbstractBugisApp(ABC):
async def __call__(self,
scope: ASGIHTTPScope | WebSocketScope | LifespanScope,
receive: Callable[[], Awaitable[Any]],
send: Callable[[Mapping[str, Any]], Awaitable[None]]) -> None:
loop = get_running_loop()
if scope['type'] == 'lifespan':
while True:
message = await receive()
if message['type'] == 'lifespan.startup':
self.setup(loop)
await send({'type': 'lifespan.startup.complete'})
elif message['type'] == 'lifespan.shutdown':
self.shutdown(loop)
await send({'type': 'lifespan.shutdown.complete'})
elif scope['type'] == 'http':
queue: Queue[Optional[bytes]] = Queue()
ctx = AsgiContext(scope, receive, send, AsyncQueueIterator(queue))
request_handling = loop.create_task(self.handle_request(ctx))
while True:
message = await receive()
if message['type'] == 'http.request':
Maybe.of(message['body']).filter(lambda it: len(it) > 0).if_present(queue.put_nowait)
if not message.get('more_body', False):
queue.put_nowait(None)
await request_handling
break
elif message['type'] == 'http.disconnect':
request_handling.cancel()
break
else:
raise NotImplementedError()
def setup(self, loop: AbstractEventLoop) -> None:
pass
def shutdown(self, loop: AbstractEventLoop) -> None:
pass
@abstractmethod
async def handle_request(self, ctx: HttpContext) -> None:
raise NotImplementedError()
def __rsgi_init__(self, loop: AbstractEventLoop) -> None:
self.setup(loop)
def __rsgi_del__(self, loop: AbstractEventLoop) -> None:
self.shutdown(loop)
async def __rsgi__(self, scope: RSGIHTTPScope, protocol: RSGIHTTPProtocol) -> None:
ctx = RsgiContext(scope, protocol)
await self.handle_request(ctx)
class BugisApp(AbstractBugisApp):
_tree: Tree
def __init__(self) -> None:
self._tree = Tree()
async def handle_request(self, ctx: HttpContext) -> None:
result = self._tree.get_handler(ctx.path, ctx.method)
if result is not None:
handler, captured = result
await handler.handle_request(ctx, captured)
else:
await ctx.send_empty(404)
pass
def route(self,
paths: StrOrStrings,
methods: Optional[HttpMethod | Sequence[HttpMethod]] = None,
recursive: bool = False) -> Callable[[HttpHandler], HttpHandler]:
def wrapped(handler: HttpHandler) -> HttpHandler:
nonlocal methods
nonlocal paths
_methods: Tuple[Optional[HttpMethod], ...]
if methods is None:
_methods = (None,)
elif isinstance(methods, HttpMethod):
_methods = (methods,)
else:
_methods = tuple(methods)
_paths: Tuple[str, ...]
if isinstance(paths, str):
_paths = (paths,)
else:
_paths = tuple(paths)
for method in _methods:
for path in _paths:
self._tree.register(path, method, handler, recursive)
return handler
return wrapped
def GET(self, path: str, recursive: bool = False) -> Callable[[HttpHandler], HttpHandler]:
return self.route(path, (HttpMethod.GET,), recursive)
def POST(self, path: str, recursive: bool = False) -> Callable[[HttpHandler], HttpHandler]:
return self.route(path, (HttpMethod.POST,), recursive)
def PUT(self, path: str, recursive: bool = False) -> Callable[[HttpHandler], HttpHandler]:
return self.route(path, (HttpMethod.PUT,), recursive)
def DELETE(self, path: str, recursive: bool = False) -> Callable[[HttpHandler], HttpHandler]:
return self.route(path, (HttpMethod.DELETE,), recursive)
def OPTIONS(self, path: str, recursive: bool = False) -> Callable[[HttpHandler], HttpHandler]:
return self.route(path, (HttpMethod.OPTIONS,), recursive)
def HEAD(self, path: str, recursive: bool = False) -> Callable[[HttpHandler], HttpHandler]:
return self.route(path, (HttpMethod.HEAD,), recursive)
def PATCH(self, path: str, recursive: bool = False) -> Callable[[HttpHandler], HttpHandler]:
return self.route(path, (HttpMethod.PATCH,), recursive)
src/kaya/_asgi.py
+148
View File
@@ -0,0 +1,148 @@
from typing import (
Sequence,
Tuple,
Dict,
Mapping,
Callable,
Any,
AsyncIterator,
Awaitable,
AsyncGenerator,
Optional,
List,
Iterable
)
from pwo import Maybe
from pathlib import Path
from ._http_method import HttpMethod
from ._http_context import HttpContext
from ._types import StrOrStrings
from ._types.asgi import HTTPScope
def decode_headers(headers: Iterable[Tuple[bytes, bytes]]) -> Dict[str, Sequence[str]]:
result: Dict[str, List[str]] = dict()
for key, value in headers:
key_str: str
value_str: str
if isinstance(key, bytes):
key_str = key.decode()
elif isinstance(key, str):
key_str = key
else:
raise NotImplementedError('This should never happen')
if isinstance(value, bytes):
value_str = value.decode()
elif isinstance(value, str):
value_str = value
else:
raise NotImplementedError('This should never happen')
ls = result.setdefault(key_str.lower(), list())
ls.append(value_str)
return {
k: tuple(v) for k, v in result.items()
}
def encode_headers(headers: Mapping[str, StrOrStrings]) -> Tuple[Tuple[bytes, bytes], ...]:
result = []
for key, value in headers.items():
if isinstance(value, str):
result.append((key.encode(), value.encode()))
elif isinstance(value, Sequence):
for single_value in value:
result.append((key.encode(), single_value.encode()))
return tuple(result)
class AsgiContext(HttpContext):
pathsend: bool
receive: Callable[[], Awaitable[Any]]
send: Callable[[Mapping[str, Any]], Awaitable[None]]
scheme: str
method: HttpMethod
path: str
query_string: str
headers: Mapping[str, Sequence[str]]
client: Optional[Tuple[str, int]]
server: Optional[Tuple[str, Optional[int]]]
request_body: AsyncIterator[bytes]
def __init__(self,
scope: HTTPScope,
receive: Callable[[], Awaitable[Any]],
send: Callable[[Mapping[str, Any]], Awaitable[None]],
request_body_iterator: AsyncIterator[bytes]):
self.receive = receive
self.send = send
self.pathsend = (Maybe.of_nullable(scope.get('extensions'))
.map(lambda it: it.get("http.response.pathsend"))
.is_present)
self.path = scope['path']
self.query_string = scope['query_string'].decode()
self.method = HttpMethod(scope['method'])
self.scheme = scope['scheme']
self.client = scope['client']
self.server = scope['server']
self.headers = decode_headers(scope['headers'])
self.request_body = request_body_iterator
async def stream_body(self,
status: int,
body_generator: AsyncGenerator[bytes, None],
headers: Optional[Mapping[str, StrOrStrings]] = None) -> None:
await self._send_head(status, headers)
async for chunk in body_generator:
await self.send({
'type': 'http.response.body',
'body': chunk,
'more_body': True
})
await self.send({
'type': 'http.response.body',
'body': '',
'more_body': False
})
async def send_bytes(self, status: int, body: bytes, headers: Optional[Mapping[str, StrOrStrings]] = None) -> None:
await self._send_head(status, headers)
await self.send({
'type': 'http.response.body',
'body': body,
})
async def send_str(self, status: int, body: str, headers: Optional[Mapping[str, StrOrStrings]] = None) -> None:
await self._send_head(status, headers)
await self.send({
'type': 'http.response.body',
'body': body.encode(),
})
async def _send_head(self, status: int, headers: Optional[Mapping[str, StrOrStrings]] = None) -> None:
await self.send({
'type': 'http.response.start',
'status': status,
'headers': Maybe.of_nullable(headers).map(encode_headers).or_else(tuple())
})
async def send_file(self,
status: int,
path: Path,
headers: Optional[Mapping[str, StrOrStrings]] = None) -> None:
if self.pathsend:
await self._send_head(status, headers)
await self.send({
'type': 'http.response.pathsend',
'path': path
})
else:
raise NotImplementedError()
async def send_empty(self, status: int, headers: Optional[Mapping[str, StrOrStrings]] = None) -> None:
await self._send_head(status, headers)
await self.send({
'type': 'http.response.body',
'body': '',
'more_body': False
})
src/kaya/_http_context.py
+52
View File
@@ -0,0 +1,52 @@
from typing import (
Callable,
Awaitable,
Tuple,
AsyncIterator,
AsyncGenerator,
Mapping,
Sequence,
Any,
Optional
)
from abc import ABC, abstractmethod
from pathlib import Path
from ._http_method import HttpMethod
from ._types.base import StrOrStrings
class HttpContext(ABC):
pathsend: bool
receive: Callable[[], Awaitable[Any]]
send: Callable[[Mapping[str, Any]], Awaitable[None]]
scheme: str
method: HttpMethod
path: str
query_string: str
headers: Mapping[str, Sequence[str]]
client: Optional[Tuple[str, int]]
server: Optional[Tuple[str, Optional[int]]]
request_body: AsyncIterator[bytes]
@abstractmethod
async def stream_body(self,
status: int,
body_generator: AsyncGenerator[bytes, None],
headers: Optional[Mapping[str, StrOrStrings]] = None) -> None:
pass
@abstractmethod
async def send_bytes(self, status: int, body: bytes, headers: Optional[Mapping[str, StrOrStrings]] = None) -> None:
pass
async def send_str(self, status: int, body: str, headers: Optional[Mapping[str, StrOrStrings]] = None) -> None:
await self.send_bytes(status, body.encode(), headers)
@abstractmethod
async def send_file(self, status: int, path: Path, headers: Optional[Mapping[str, StrOrStrings]] = None) -> None:
pass
@abstractmethod
async def send_empty(self, status: int, headers: Optional[Mapping[str, StrOrStrings]] = None) -> None:
pass
src/kaya/_http_method.py
+11
View File
@@ -0,0 +1,11 @@
from enum import StrEnum
class HttpMethod(StrEnum):
OPTIONS = 'OPTIONS'
HEAD = 'HEAD'
GET = 'GET'
POST = 'POST'
PUT = 'PUT'
DELETE = 'DELETE'
PATCH = 'PATCH'
src/kaya/_node.py
+12
View File
@@ -0,0 +1,12 @@
from dataclasses import dataclass
from typing import (
Optional,
Dict,
List,
)
from ._types import NodeType
from ._path_handler import PathHandler
from ._path_matcher import PathMatcher
src/kaya/_path_handler.py
+33
View File
@@ -0,0 +1,33 @@
from abc import ABC, abstractmethod
from typing import (
Sequence,
Dict,
Optional
)
from dataclasses import dataclass, field
from ._http_context import HttpContext
@dataclass
class Matches:
kwargs: Dict[str, str] = field(default_factory=dict)
path: Optional[Sequence[str]] = None
unmatched_paths: Sequence[str] = field(default_factory=list)
class PathHandler(ABC):
@abstractmethod
async def handle_request(self, ctx: HttpContext, captured: Matches) -> None:
pass
@property
@abstractmethod
def recursive(self) -> bool:
raise NotImplementedError()
type PathHandlers = (PathHandler | Sequence[PathHandler])
src/kaya/_path_matcher.py
+100
View File
@@ -0,0 +1,100 @@
from fnmatch import fnmatch
from abc import ABC, abstractmethod
from typing import Optional, Sequence, Dict, List, Union, Set
from dataclasses import dataclass, field
from ._path_handler import PathHandler
from ._http_method import HttpMethod
from ._types import NodeType, PathMatcherResult
@dataclass
class Node:
key: NodeType
parent: Optional[Union['Node', 'PathMatcher']]
children: Dict[NodeType, 'Node']
handlers: List[PathHandler]
path_matchers: List['PathMatcher']
supported_methods: Set[Optional[HttpMethod]] = field(default_factory=set)
class PathMatcher(ABC):
parent: Optional[Union['Node', 'PathMatcher']]
children: Dict[NodeType, Node]
handlers: List[PathHandler]
path_matchers: List['PathMatcher']
def __init__(self,
parent: Optional[Union['Node', 'PathMatcher']],
children: Dict[NodeType, Node],
handlers: List[PathHandler],
path_matchers: List['PathMatcher']
):
self.parent = parent
self.children = children
self.handlers = handlers
self.path_matchers = path_matchers
self.supported_methods: Set[Optional[HttpMethod]] = set()
@abstractmethod
def match(self, path: Sequence[str]) -> Optional[PathMatcherResult]:
pass
class StrMatcher(PathMatcher):
name: str
def __init__(self,
name: str,
parent: Optional[Node | PathMatcher],
children: Dict[NodeType, Node],
handlers: List[PathHandler],
path_matchers: List[PathMatcher],
):
super().__init__(parent, children, handlers, path_matchers)
self.name = name
def match(self, path: Sequence[str]) -> Optional[PathMatcherResult]:
if len(path):
return {self.name: path[0]}
else:
return None
class IntMatcher(PathMatcher):
name: str
def __init__(self,
name: str,
parent: Optional[Node | PathMatcher],
children: Dict[NodeType, Node],
handlers: List[PathHandler],
path_matchers: List[PathMatcher],
):
super().__init__(parent, children, handlers, path_matchers)
self.name = name
def match(self, path: Sequence[str]) -> Optional[PathMatcherResult]:
if len(path) > 0:
try:
return {self.name: int(path[0])}
except ValueError:
return None
else:
return None
class GlobMatcher(PathMatcher):
pattern: str
def __init__(self,
pattern: str,
parent: Optional[Node | PathMatcher],
children: Dict[NodeType, Node],
handlers: List[PathHandler],
path_matchers: List[PathMatcher],
):
super().__init__(parent, children, handlers, path_matchers)
self.pattern = pattern
def match(self, path: Sequence[str]) -> Optional[PathMatcherResult]:
return path if fnmatch('/'.join(path), self.pattern) else None
src/kaya/_rsgi.py
+107
View File
@@ -0,0 +1,107 @@
from functools import reduce
from pathlib import Path
from typing import (
Any,
Sequence,
Mapping,
AsyncIterator,
Tuple,
AsyncGenerator,
Optional,
List,
Dict,
Callable,
cast
)
from granian._granian import RSGIHTTPProtocol, RSGIHTTPScope # type: ignore[attr-defined]
from pwo import Maybe
from ._types import StrOrStrings
from ._http_context import HttpContext
from ._http_method import HttpMethod
class RsgiContext(HttpContext):
protocol: RSGIHTTPProtocol
scheme: str
method: HttpMethod
path: str
query_string: str
headers: Mapping[str, Sequence[str]]
client: Optional[Tuple[str, int]]
server: Optional[Tuple[str, Optional[int]]]
request_body: AsyncIterator[bytes]
head = Optional[Tuple[int, Sequence[Tuple[str, str]]]]
def __init__(self, scope: RSGIHTTPScope, protocol: RSGIHTTPProtocol):
self.scheme = scope.scheme
self.path = scope.path
self.method = HttpMethod(scope.method)
self.query_string = scope.query_string
def acc(d: Dict[str, List[str]], t: Tuple[str, str]) -> Dict[str, List[str]]:
d.setdefault(t[0].lower(), list()).append(t[1])
return d
fun = cast(Callable[[Mapping[str, Sequence[str]], tuple[str, str]], Mapping[str, Sequence[str]]], acc)
self.headers = reduce(fun, scope.headers.items(), {})
self.client = (Maybe.of(scope.client.split(':'))
.map(lambda it: (it[0], int(it[1])))
.or_else_throw(RuntimeError))
self.server = (Maybe.of(scope.server.split(':'))
.map(lambda it: (it[0], int(it[1])))
.or_else_throw(RuntimeError))
self.request_body = cast(AsyncIterator[bytes], protocol)
self.protocol = protocol
# @staticmethod
# def _rearrange_headers(headers: Mapping[str, Sequence[str]]) -> List[Tuple[str, str]]:
# return list(
# ((key, value) for key, values in headers.items() for value in values)
# )
@staticmethod
def _rearrange_headers(headers: Mapping[str, StrOrStrings]) -> List[Tuple[str, str]]:
result = []
for key, value in headers.items():
if isinstance(value, str):
result.append((key, value))
elif isinstance(value, Sequence):
for single_value in value:
result.append((key, single_value))
return result
async def stream_body(self,
status: int,
body_generator: AsyncGenerator[bytes, None],
headers: Optional[Mapping[str, StrOrStrings]] = None) -> None:
transport = self.protocol.response_stream(status,
Maybe.of_nullable(headers)
.map(self._rearrange_headers)
.or_else([]))
async for chunk in body_generator:
await transport.send_bytes(chunk)
async def send_bytes(self, status: int, body: bytes, headers: Optional[Mapping[str, StrOrStrings]] = None) -> None:
rearranged_headers = Maybe.of_nullable(headers).map(RsgiContext._rearrange_headers).or_else(list())
if len(body) > 0:
self.protocol.response_bytes(status, rearranged_headers, body)
else:
self.protocol.response_empty(status, rearranged_headers)
async def send_str(self, status: int, body: str, headers: Optional[Mapping[str, StrOrStrings]] = None) -> None:
rearranged_headers = Maybe.of_nullable(headers).map(RsgiContext._rearrange_headers).or_else(list())
if len(body) > 0:
self.protocol.response_str(status, rearranged_headers, body)
else:
self.protocol.response_empty(status, rearranged_headers)
async def send_file(self, status: int, path: Path, headers: Optional[Mapping[str, StrOrStrings]] = None) -> None:
rearranged_headers = (Maybe.of_nullable(headers).map(RsgiContext._rearrange_headers)
.or_else(list()))
self.protocol.response_file(status, rearranged_headers, str(path))
async def send_empty(self, status: int, headers: Optional[Mapping[str, StrOrStrings]] = None) -> None:
rearranged_headers = Maybe.of_nullable(headers).map(RsgiContext._rearrange_headers).or_else(list())
self.protocol.response_empty(status, rearranged_headers)
src/kaya/_tree.py
+262
View File
@@ -0,0 +1,262 @@
from itertools import chain
from typing import (
Sequence,
Awaitable,
Callable,
Optional,
Generator,
Self,
List,
Tuple,
Mapping,
Any,
)
from typing_extensions import Unpack
from urllib.parse import urlparse
from pwo import Maybe, index_of_with_escape
from ._http_context import HttpContext
from ._http_method import HttpMethod
from ._path_handler import PathHandler
from ._path_matcher import PathMatcher, IntMatcher, GlobMatcher, StrMatcher, Node
from ._path_handler import Matches
from ._types import NodeType
class Tree:
def __init__(self) -> None:
self.root = Node('/', None, {}, [], [])
def search(self, path: Generator[str, None, None], method: HttpMethod) \
-> Optional[Tuple[Node | PathMatcher, Matches]]:
paths: List[str] = list(path)
result: Node | PathMatcher = self.root
matches = Matches()
it, i = iter((it for it in paths)), -1
while True:
node = result
leaf, i = next(it, None), i + 1
if leaf is None:
break
child = node.children.get(leaf)
if child is None and isinstance(leaf, str):
specific_matchers: List[PathMatcher] = []
generic_matchers: List[PathMatcher] = []
for matcher in node.path_matchers:
if not self._supports_method(matcher, method):
continue
if None in matcher.supported_methods:
generic_matchers.append(matcher)
else:
specific_matchers.append(matcher)
for matcher in specific_matchers + generic_matchers:
match = matcher.match(paths[i:])
if match is not None:
if isinstance(match, Mapping):
matches.kwargs.update(match)
elif isinstance(match, Sequence):
matches.path = match
result = matcher
break
else:
break
else:
result = child
child = result.children.get(method)
if child is not None:
result = child
matches.unmatched_paths = paths[i:]
return None if result == self.root else (result, matches)
def add(self, path: Generator[str, None, None], method: Optional[HttpMethod], *path_handlers: PathHandler) -> Node | PathMatcher:
lineage: Generator[NodeType, None, None] = (it for it in
chain(path,
Maybe.of_nullable(method)
.map(lambda it: [it])
.or_else([])))
result: Node | PathMatcher = self.root
it = iter(lineage)
while True:
node = result
leaf = next(it, None)
if leaf is None:
break
child = node.children.get(leaf)
if child is None:
break
else:
result = child
key = leaf
while key is not None:
new_node = self.parse(key, result)
if isinstance(new_node, Node):
result.children[key] = new_node
else:
self._check_matcher_conflict(result, method)
result.path_matchers.append(new_node)
result = new_node
key = next(it, None)
result.handlers = list(chain(result.handlers, path_handlers))
self._add_supported_method(result, method)
return result
@staticmethod
def _supports_method(node: Node | PathMatcher, method: HttpMethod) -> bool:
return None in node.supported_methods or method in node.supported_methods
def _check_matcher_conflict(self, node: Node | PathMatcher, method: Optional[HttpMethod]) -> None:
new_is_generic = method is None
for existing in node.path_matchers:
existing_is_generic = None in existing.supported_methods
if new_is_generic and existing_is_generic:
raise ValueError(
"Cannot register two method-agnostic path matchers at the same node"
)
if not new_is_generic and not existing_is_generic:
if method in existing.supported_methods:
raise ValueError(
f"Cannot register path matcher because it overlaps on method {method} "
f"with an existing matcher at the same node"
)
def _add_supported_method(self, node: Node | PathMatcher, method: Optional[HttpMethod]) -> None:
current: Optional[Node | PathMatcher] = node
while current is not None:
if method in current.supported_methods:
break
current.supported_methods.add(method)
current = current.parent
def register(self,
path: str,
method: Optional[HttpMethod],
callback: Callable[[HttpContext, Unpack[Any]], Awaitable[None]],
recursive: bool) -> None:
class Handler(PathHandler):
async def handle_request(self, ctx: HttpContext, captured: Matches) -> None:
args = Maybe.of_nullable(captured.path).map(lambda it: [it]).or_else([])
await callback(ctx, *args, **captured.kwargs)
@property
def recursive(self) -> bool:
return recursive
handler = Handler()
self.add((p for p in PathIterator(path)), method, handler)
def find_node(self, path: Generator[str, None, None], method: HttpMethod = HttpMethod.GET) \
-> Optional[Tuple[Node | PathMatcher, Matches]]:
return (Maybe.of_nullable(self.search(path, method))
.filter(lambda it: len(it[0].handlers) > 0)
.or_none())
def get_handler(self, url: str, method: HttpMethod = HttpMethod.GET) \
-> Optional[Tuple[PathHandler, Matches]]:
path = urlparse(url).path
result: Optional[Tuple[Node | PathMatcher, Matches]] = self.find_node((p for p in PathIterator(path)), method)
if result is None:
return None
node, captured = result
# requested = (p for p in PathIterator(path))
# found = reversed([n for n in NodeAncestryIterator(node) if n != self.root])
# unmatched: List[str] = []
# for r, f in zip(requested, found):
# if f is None:
# unmatched.append(r)
for handler in node.handlers:
if len(captured.unmatched_paths) == 0:
return handler, captured
elif handler.recursive:
return handler, captured
# if handler.match(unmatched, method):
# return (handler, unmatched)
return None
def parse(self, leaf: str, parent: Optional[Node | PathMatcher]) -> Node | PathMatcher:
start = 0
result = index_of_with_escape(leaf, '${', '\\', 0)
if result >= 0:
start = result + 2
end = leaf.index('}', start + 2)
definition = leaf[start:end]
try:
colon = definition.index(':')
except ValueError:
colon = None
if colon is None:
key = definition
kind = 'str'
else:
key = definition[:colon]
kind = definition[colon+1:] if colon is not None else 'str'
if kind == 'str':
return StrMatcher(name=key, parent=parent, children={}, handlers=[], path_matchers=[])
elif kind == 'int':
return IntMatcher(name=key, parent=parent, children={}, handlers=[], path_matchers=[])
else:
raise ValueError(f"Unknown kind: '{kind}'")
result = index_of_with_escape(leaf, '*', '\\', 0)
if result >= 0:
return GlobMatcher(pattern=leaf, parent=parent, children={}, handlers=[], path_matchers=[])
else:
return Node(key=leaf, parent=parent, children={}, handlers=[], path_matchers=[])
class PathIterator:
path: str
cursor: int
def __init__(self, path: str):
self.path = path
self.cursor = 0
def __iter__(self) -> Self:
return self
def advance_cursor(self, next_value: int) -> None:
if next_value < len(self.path):
self.cursor = next_value
else:
self.cursor = -1
def __next__(self) -> str:
if self.cursor < 0:
raise StopIteration()
else:
while self.cursor >= 0:
next_separator = self.path.find('/', self.cursor)
if next_separator < 0:
result = self.path[self.cursor:]
self.cursor = next_separator
return result
elif next_separator == self.cursor:
self.advance_cursor(next_separator + 1)
else:
result = self.path[self.cursor:next_separator]
self.advance_cursor(next_separator + 1)
return result
raise StopIteration()
class NodeAncestryIterator:
node: Node | PathMatcher
def __init__(self, node: Node):
self.node = node
def __iter__(self) -> Self:
return self
def __next__(self) -> Node | PathMatcher:
parent = self.node.parent
if parent is None:
raise StopIteration()
else:
self.node = parent
return parent
src/kaya/_types/__init__.py
+92
View File
@@ -0,0 +1,92 @@
from typing import (
TypedDict,
Literal,
Iterable,
Tuple,
Optional,
NotRequired,
Dict,
Any,
Union,
Mapping,
Sequence
)
from .base import StrOrStrings, PathMatcherResult
from kaya._http_method import HttpMethod
type NodeType = (str | HttpMethod)
class ASGIVersions(TypedDict):
spec_version: str
version: Union[Literal["2.0"], Literal["3.0"]]
class HTTPScope(TypedDict):
type: Literal["http"]
asgi: ASGIVersions
http_version: str
method: str
scheme: str
path: str
raw_path: bytes
query_string: bytes
root_path: str
headers: Iterable[Tuple[bytes, bytes]]
client: Optional[Tuple[str, int]]
server: Optional[Tuple[str, Optional[int]]]
state: NotRequired[Dict[str, Any]]
extensions: Optional[Dict[str, Dict[object, object]]]
class WebSocketScope(TypedDict):
type: Literal["websocket"]
asgi: ASGIVersions
http_version: str
scheme: str
path: str
raw_path: bytes
query_string: bytes
root_path: str
headers: Iterable[Tuple[bytes, bytes]]
client: Optional[Tuple[str, int]]
server: Optional[Tuple[str, Optional[int]]]
subprotocols: Iterable[str]
state: NotRequired[Dict[str, Any]]
extensions: Optional[Dict[str, Dict[object, object]]]
class LifespanScope(TypedDict):
type: Literal["lifespan"]
asgi: ASGIVersions
state: NotRequired[Dict[str, Any]]
class RSGI:
class Scope(TypedDict):
proto: Literal['http'] # = 'http'
rsgi_version: str
http_version: str
server: str
client: str
scheme: str
method: str
path: str
query_string: str
headers: Mapping[str, str]
authority: Optional[str]
__all__ = [
'HttpMethod',
'HTTPScope',
'LifespanScope',
'RSGI',
'ASGIVersions',
'WebSocketScope',
'NodeType',
'StrOrStrings',
'PathMatcherResult'
]
src/kaya/_types/asgi.py
+57
View File
@@ -0,0 +1,57 @@
from typing import (
Sequence,
TypedDict,
Literal,
Iterable,
Tuple,
Optional,
NotRequired,
Dict,
Any,
Union
)
class ASGIVersions(TypedDict):
spec_version: str
version: Union[Literal["2.0"], Literal["3.0"]]
class HTTPScope(TypedDict):
type: Literal["http"]
asgi: ASGIVersions
http_version: str
method: str
scheme: str
path: str
raw_path: bytes
query_string: bytes
root_path: str
headers: Iterable[Tuple[bytes, bytes]]
client: Optional[Tuple[str, int]]
server: Optional[Tuple[str, Optional[int]]]
state: NotRequired[Dict[str, Any]]
extensions: Optional[Dict[str, Dict[object, object]]]
class WebSocketScope(TypedDict):
type: Literal["websocket"]
asgi: ASGIVersions
http_version: str
scheme: str
path: str
raw_path: bytes
query_string: bytes
root_path: str
headers: Iterable[Tuple[bytes, bytes]]
client: Optional[Tuple[str, int]]
server: Optional[Tuple[str, Optional[int]]]
subprotocols: Iterable[str]
state: NotRequired[Dict[str, Any]]
extensions: Optional[Dict[str, Dict[object, object]]]
class LifespanScope(TypedDict):
type: Literal["lifespan"]
asgi: ASGIVersions
state: NotRequired[Dict[str, Any]]
src/kaya/_types/base.py
+4
View File
@@ -0,0 +1,4 @@
from typing import Sequence, Mapping, Any
type StrOrStrings = (str | Sequence[str])
type PathMatcherResult = Mapping[str, Any] | Sequence[str]
src/kaya/_types/rsgi.py
+26
View File
@@ -0,0 +1,26 @@
from typing import (
Sequence,
TypedDict,
Literal,
Iterable,
Tuple,
Optional,
NotRequired,
Dict,
Any,
Union,
Mapping,
)
class HTTPScope(TypedDict):
proto: Literal['http']
rsgi_version: str
http_version: str
server: str
client: str
scheme: str
method: str
path: str
query_string: str
headers: Mapping[str, str]
authority: Optional[str]
src/kaya/py.typed
View File
tests/test_asgi.py
+192
View File
@@ -0,0 +1,192 @@
import unittest
import json
import httpx
from pwo import async_test
from kaya import BugisApp, HttpContext, HttpMethod
from typing import Sequence, List
class AsgiTest(unittest.TestCase):
app: BugisApp
def setUp(self):
self.app = BugisApp()
@self.app.GET('/hello')
@self.app.GET('/hello2')
@self.app.route('/hello3')
@self.app.GET('/hello/*')
async def handle_request(ctx: HttpContext) -> None:
async for chunk in ctx.request_body:
print(chunk)
await ctx.send_str(200, 'Hello World!')
@self.app.route(('/foo/bar',), HttpMethod.PUT, recursive=True)
async def handle_request(ctx: HttpContext) -> None:
async for chunk in ctx.request_body:
print(chunk)
await ctx.send_str(200, ctx.path)
@self.app.route(('/foo/*',), HttpMethod.PUT, recursive=True)
async def handle_request(ctx: HttpContext, path: Sequence[str]) -> None:
async for chunk in ctx.request_body:
print(chunk)
await ctx.send_str(200, json.dumps(path))
@self.app.route(('/foo/*',), recursive=True)
async def handle_request(ctx: HttpContext, path: Sequence[str]) -> None:
await ctx.send_str(500, f"Unable to handle request for {ctx.path}")
@self.app.GET('/employee/${employee_id}')
async def handle_request(ctx: HttpContext, employee_id: str) -> None:
async for chunk in ctx.request_body:
print(chunk)
await ctx.send_str(200, employee_id)
@self.app.GET('/square/${x:int}')
async def handle_request(ctx: HttpContext, x: int) -> None:
async for chunk in ctx.request_body:
print(chunk)
await ctx.send_str(200, str(x * x))
@self.app.GET('/department/${department_id:int}/employee/${employee_id:int}')
async def handle_request(ctx: HttpContext, department_id: int, employee_id: int) -> None:
async for chunk in ctx.request_body:
print(chunk)
await ctx.send_str(200, json.dumps({
'department_id': department_id,
'employee_id': employee_id
}))
@self.app.PUT('/hello/*', recursive=True)
async def handle_request(ctx: HttpContext, _: List[str]) -> None:
await ctx.stream_body(200, (chunk async for chunk in ctx.request_body))
@async_test
async def test_hello(self):
transport = httpx.ASGITransport(app=self.app)
async with httpx.AsyncClient(transport=transport, base_url="http://127.0.0.1:80") as client:
r = await client.get("/hello")
self.assertEqual(200, r.status_code)
self.assertEqual(r.text, "Hello World!")
r = await client.get("/hello2")
self.assertEqual(200, r.status_code)
self.assertEqual(r.text, "Hello World!")
r = await client.post("/hello3")
self.assertEqual(200, r.status_code)
self.assertEqual(r.text, "Hello World!")
r = await client.get("/hello4")
self.assertEqual(404, r.status_code)
self.assertTrue(len(r.text) == 0)
body = {'name': 'John', 'surname': 'Smith'}
r = await client.put("/hello/foo/bar", json=body)
self.assertEqual(200, r.status_code)
ans = json.loads(r.text)
self.assertTrue(body, ans)
@async_test
async def test_foo(self):
transport = httpx.ASGITransport(app=self.app)
async with httpx.AsyncClient(transport=transport, base_url="http://127.0.0.1:80") as client:
r = await client.put("/foo/fizz/baz")
self.assertEqual(200, r.status_code)
response = json.loads(r.text)
self.assertEqual(['fizz', 'baz'], response)
r = await client.get("/foo/not-put")
self.assertEqual(500, r.status_code)
self.assertEqual("Unable to handle request for /foo/not-put", r.text)
@async_test
async def test_method_agnostic_fallback_order_independence(self):
app = BugisApp()
@app.route(('/foo/*',), recursive=True)
async def handle_request(ctx: HttpContext, path: Sequence[str]) -> None:
await ctx.send_str(500, f"Unable to handle request for {ctx.path}")
@app.route(('/foo/*',), HttpMethod.PUT, recursive=True)
async def handle_request(ctx: HttpContext, path: Sequence[str]) -> None:
await ctx.send_str(200, json.dumps(path))
transport = httpx.ASGITransport(app=app)
async with httpx.AsyncClient(transport=transport, base_url="http://127.0.0.1:80") as client:
r = await client.put("/foo/fizz/baz")
self.assertEqual(200, r.status_code)
self.assertEqual(['fizz', 'baz'], json.loads(r.text))
r = await client.get("/foo/not-put")
self.assertEqual(500, r.status_code)
self.assertEqual("Unable to handle request for /foo/not-put", r.text)
@async_test
async def test_disjoint_method_specific_matchers(self):
app = BugisApp()
@app.route(('/foo/*',), HttpMethod.PUT, recursive=True)
async def handle_request(ctx: HttpContext, path: Sequence[str]) -> None:
await ctx.send_str(200, "PUT")
@app.route(('/foo/*',), HttpMethod.GET, recursive=True)
async def handle_request(ctx: HttpContext, path: Sequence[str]) -> None:
await ctx.send_str(200, "GET")
transport = httpx.ASGITransport(app=app)
async with httpx.AsyncClient(transport=transport, base_url="http://127.0.0.1:80") as client:
r = await client.put("/foo/bar")
self.assertEqual(200, r.status_code)
self.assertEqual("PUT", r.text)
r = await client.get("/foo/bar")
self.assertEqual(200, r.status_code)
self.assertEqual("GET", r.text)
@async_test
async def test_foo_bar(self):
transport = httpx.ASGITransport(app=self.app)
async with httpx.AsyncClient(transport=transport, base_url="http://127.0.0.1:80") as client:
r = await client.put("/foo/bar/baz")
self.assertEqual(200, r.status_code)
self.assertEqual('/foo/bar/baz', r.text)
@async_test
async def test_employee(self):
transport = httpx.ASGITransport(app=self.app)
async with httpx.AsyncClient(transport=transport, base_url="http://127.0.0.1:80") as client:
r = await client.get("/employee/101325")
self.assertEqual(200, r.status_code)
self.assertEqual(r.text, '101325')
@async_test
async def test_square(self):
transport = httpx.ASGITransport(app=self.app)
async with httpx.AsyncClient(transport=transport, base_url="http://127.0.0.1:80") as client:
x = 30
r = await client.get(f"/square/{x}")
self.assertEqual(200, r.status_code)
self.assertEqual(r.text, str(x * x))
@async_test
async def test_department_employee(self):
transport = httpx.ASGITransport(app=self.app)
async with httpx.AsyncClient(transport=transport, base_url="http://127.0.0.1:80") as client:
r = await client.get("department/189350/employee/101325")
self.assertEqual(200, r.status_code)
response = json.loads(r.text)
self.assertEqual({
'department_id': 189350,
'employee_id': 101325
}, response)
tests/test_tree.py
+104
View File
@@ -0,0 +1,104 @@
from typing import Sequence, Tuple, Optional, List
from kaya import Tree, PathHandler, HttpContext, HttpMethod, PathIterator
from kaya import HttpMethod
from pwo import Maybe
import unittest
class PathIteratorTest(unittest.TestCase):
cases: Tuple[Tuple[str, Tuple[str, ...]], ...] = (
('/', tuple()),
('root/foo', ('root', 'foo')),
('/root', ('root',)),
('/root', ('root',)),
('/root/', ('root',)),
('/root/bar/', ('root', 'bar')),
)
def test_path_iterator(self):
for (case, expected) in self.cases:
with self.subTest(case) as _:
components = tuple((c for c in PathIterator(case)))
self.assertEqual(expected, components)
class TreeTest(unittest.TestCase):
tree: Tree
handlers: List[PathHandler]
def setUp(self):
self.tree = Tree()
class TestHandler(PathHandler):
def handle_request(self, ctx: HttpContext):
pass
@property
def recursive(self) -> bool:
return True
self.handlers = [TestHandler() for _ in range(20)]
routes: Tuple[Tuple[Tuple[str, ...], Optional[HttpMethod], PathHandler], ...] = (
(('home', 'something'), HttpMethod.GET, self.handlers[0]),
(('home', 'something_else'), HttpMethod.GET, self.handlers[1]),
(('home', 'something_else'), HttpMethod.POST, self.handlers[2]),
(('home', 'something', 'object'), HttpMethod.GET, self.handlers[3]),
(('home', 'something_else', 'foo'), HttpMethod.GET, self.handlers[4]),
(('home',), HttpMethod.GET, self.handlers[5]),
(('home',), HttpMethod.POST, self.handlers[6]),
(('home',), None, self.handlers[7]),
(('home', '*.md'), None, self.handlers[8]),
(('home', 'something', '*', 'blah', '*.md'), None, self.handlers[9]),
(('home', 'bar', '*'), None, self.handlers[10]),
)
for path, method, handler in routes:
self.tree.add((p for p in path), method, handler)
def test_tree(self):
cases: Tuple[Tuple[str, HttpMethod, Optional[int]], ...] = (
('http://localhost:127.0.0.1:5432/home/something', HttpMethod.GET, 0),
('http://localhost:127.0.0.1:5432/home/something_else', HttpMethod.GET, 1),
('http://localhost:127.0.0.1:5432/home/something_else', HttpMethod.POST, 2),
('http://localhost:127.0.0.1:5432/home/something/object', HttpMethod.GET, 3),
('http://localhost:127.0.0.1:5432/home/something_else/foo', HttpMethod.GET, 4),
('http://localhost:127.0.0.1:5432/', HttpMethod.GET, None),
('http://localhost:127.0.0.1:5432/home', HttpMethod.GET, 5),
('http://localhost:127.0.0.1:5432/home', HttpMethod.POST, 6),
('http://localhost:127.0.0.1:5432/home', HttpMethod.PUT, 7),
('http://localhost:127.0.0.1:5432/home/README.md', HttpMethod.GET, 8),
('http://localhost:127.0.0.1:5432/home/something/ciao/blah/README.md', HttpMethod.GET, 9),
('http://localhost:127.0.0.1:5432/home/bar/ciao/blah/README.md', HttpMethod.GET, 10),
)
for url, method, handler_num in cases:
with self.subTest(f"{str(method)} {url}"):
res = self.tree.get_handler(url, method)
self.assertIs(Maybe.of(handler_num).map(self.handlers.__getitem__).or_none(),
Maybe.of_nullable(res).map(lambda it: it[0]).or_none())
def test_two_method_agnostic_matchers_raise(self):
tree = Tree()
tree.add((p for p in ('foo', '*')), None, self.handlers[0])
with self.assertRaises(ValueError):
tree.add((p for p in ('foo', '*')), None, self.handlers[1])
def test_two_overlapping_method_specific_matchers_raise(self):
tree = Tree()
tree.add((p for p in ('foo', '${id:int}')), HttpMethod.PUT, self.handlers[0])
with self.assertRaises(ValueError):
tree.add((p for p in ('foo', '${name:str}')), HttpMethod.PUT, self.handlers[1])
def test_disjoint_method_specific_matchers_allowed(self):
tree = Tree()
tree.add((p for p in ('foo', '*')), HttpMethod.PUT, self.handlers[0])
tree.add((p for p in ('foo', '*')), HttpMethod.GET, self.handlers[1])
put_handler = Maybe.of_nullable(tree.get_handler('/foo/bar', HttpMethod.PUT)).map(lambda it: it[0]).or_none()
get_handler = Maybe.of_nullable(tree.get_handler('/foo/bar', HttpMethod.GET)).map(lambda it: it[0]).or_none()
self.assertIs(self.handlers[0], put_handler)
self.assertIs(self.handlers[1], get_handler)