Architecture

This document explains how Teyaotlani is structured internally, helping you understand the codebase for contributions or customization.

Module organization

Teyaotlani is organized into focused modules:

teyaotlani/
├── __init__.py       # Public API exports
├── __main__.py       # CLI entry point
├── client/           # Client implementation
│   ├── protocol.py   # Client protocol handler
│   └── session.py    # SpartanClient class
├── server/           # Server implementation
│   ├── config.py     # ServerConfig class
│   ├── handler.py    # Request handlers
│   ├── middleware.py # Rate limiting, access control
│   ├── protocol.py   # Server protocol handler
│   └── server.py     # Server runner
├── protocol/         # Shared protocol components
│   ├── constants.py  # Protocol constants
│   ├── request.py    # SpartanRequest class
│   ├── response.py   # SpartanResponse class
│   └── status.py     # StatusCode enum
├── content/          # Content handling
│   └── gemtext.py    # Gemtext utilities
└── utils/            # Shared utilities
    └── url.py        # URL parsing

Design principles

Async-first

Teyaotlani is built entirely on Python's asyncio. This provides:

• High concurrency - Handle many connections efficiently
• Non-blocking I/O - No threads needed for I/O operations
• Native coroutines - Clean, readable async code

# All public APIs are async
async with SpartanClient() as client:
    response = await client.get("spartan://example.com/")

Protocol-oriented

Both client and server use asyncio.Protocol for network communication:

class SpartanClientProtocol(asyncio.Protocol):
    def connection_made(self, transport):
        # Send request
        ...

    def data_received(self, data):
        # Parse response
        ...

Benefits:

• Direct control over the connection lifecycle
• Efficient buffer management
• Clean separation of concerns

Context managers

Resources that need cleanup use context managers:

async with SpartanClient() as client:
    # Client is properly initialized
    response = await client.get(url)
# Client is properly closed

This ensures connections are always cleaned up, even on errors.

Client architecture

SpartanClient

The main client class manages connections:

SpartanClient
├── get(url) -> SpartanResponse
├── upload(url, data) -> SpartanResponse
└── Internal:
    ├── _connect(host, port)
    └── _send_request(request)

Request flow

1. Parse URL to extract host, port, path
2. Create SpartanRequest object
3. Open connection to server
4. Send request bytes
5. Receive and parse response
6. Return SpartanResponse

# Simplified flow
request = SpartanRequest(host, path, content_length=0)
transport, protocol = await loop.create_connection(...)
transport.write(request.to_bytes())
response = await protocol.response_future

Server architecture

Components

Server
├── ServerConfig      # Configuration container
├── SpartanServer     # Main server class
├── SpartanProtocol   # Per-connection handler
├── RequestHandler    # Static file serving
└── Middleware
    ├── RateLimiter
    └── AccessControl

Request handling flow

1. Accept connection
2. Check access control (if enabled)
3. Parse request
4. Check rate limits (if enabled)
5. Route to handler
6. Send response

Handler chain

Handlers process requests in order:

handlers = [
    StaticFileHandler(config),
    UploadHandler(config),
]

for handler in handlers:
    response = await handler.handle(request)
    if response:
        return response

Middleware

Middleware runs before handlers:

# Rate limiting
if rate_limiter.is_limited(client_ip):
    return Response(5, "Rate limited")

# Access control
if not access_control.is_allowed(client_ip):
    return Response(4, "Access denied")

Protocol layer

SpartanRequest

Represents an incoming request:

@dataclass
class SpartanRequest:
    host: str
    path: str
    content_length: int
    data: bytes = b""

SpartanResponse

Represents a server response:

@dataclass
class SpartanResponse:
    status: int
    meta: str
    body: str | bytes = ""

StatusCode

Enum for protocol status codes:

class StatusCode(IntEnum):
    SUCCESS = 2
    REDIRECT = 3
    CLIENT_ERROR = 4
    SERVER_ERROR = 5

Configuration system

ServerConfig uses a dataclass with TOML loading:

@dataclass
class ServerConfig:
    host: str = "localhost"
    port: int = 300
    document_root: Path = Path("./capsule")
    # ... more fields

    @classmethod
    def from_toml(cls, path: str) -> "ServerConfig":
        data = tomllib.load(open(path, "rb"))
        return cls(**data)

Content handling

Gemtext utilities

The content.gemtext module provides Gemtext parsing:

from teyaotlani.content.gemtext import (
    has_input_prompt,
    extract_input_prompts,
)

Extension points

Custom handlers

Create handlers by implementing the handler interface:

class MyHandler:
    async def handle(self, request: SpartanRequest) -> SpartanResponse | None:
        if request.path.startswith("/api/"):
            return SpartanResponse(2, "application/json", '{"ok": true}')
        return None  # Pass to next handler

Middleware

Add middleware for cross-cutting concerns:

class LoggingMiddleware:
    async def process(self, request: SpartanRequest) -> SpartanRequest:
        logger.info(f"Request: {request.path}")
        return request

Testing

Tests are organized by module:

tests/
├── test_client/      # Client tests
├── test_server/      # Server tests
├── test_protocol/    # Protocol tests
├── test_integration/ # End-to-end tests
└── conftest.py       # Shared fixtures

Run tests with:

uv run pytest

Contributing

When contributing:

1. Follow existing patterns
2. Add tests for new functionality
3. Update documentation
4. Run uv run ruff check and uv run ruff format