Server API Reference

The server module provides functionality for running Spartan protocol servers.

Quick example

import asyncio
from teyaotlani import ServerConfig, run_server

async def main():
    config = ServerConfig(
        host="localhost",
        port=3000,
        document_root="./capsule",
    )
    await run_server(config)

asyncio.run(main())

ServerConfig

teyaotlani.ServerConfig dataclass

ServerConfig(
    host="localhost",
    port=DEFAULT_PORT,
    document_root=".",
    max_file_size=DEFAULT_MAX_FILE_SIZE,
    enable_directory_listing=False,
    index_files=(lambda: ["index.gmi", "index.gemini"])(),
    enable_rate_limiting=True,
    rate_limit_capacity=10,
    rate_limit_refill_rate=1.0,
    rate_limit_retry_after=30,
    enable_access_control=False,
    access_control_allow_list=None,
    access_control_deny_list=None,
    access_control_default_allow=True,
    enable_upload=False,
    upload_dir=None,
    max_upload_size=10 * 1024 * 1024,
    enable_delete=False,
    hash_client_ips=True,
    log_level="INFO",
    json_logs=False,
)

Configuration for Spartan server.

Attributes:

Name	Type	Description
host	str	Server host address.
port	int	Server port (default: 300).
document_root	Path | str	Path to directory containing files to serve.

Examples:

>>> config = ServerConfig(
...     host="localhost",
...     port=300,
...     document_root=Path("./capsule"),
... )

Functions

__post_init__

__post_init__()

Validate and normalize configuration.

Source code in src/teyaotlani/server/config.py

def __post_init__(self) -> None:
    """Validate and normalize configuration."""
    # Convert string paths to Path objects
    if isinstance(self.document_root, str):
        self.document_root = Path(self.document_root)

    if isinstance(self.upload_dir, str):
        self.upload_dir = Path(self.upload_dir)

    # Resolve document root
    self.document_root = self.document_root.resolve()

validate

validate()

Validate configuration at runtime.

Raises:

Type	Description
ValueError	If configuration is invalid.

Source code in src/teyaotlani/server/config.py

def validate(self) -> None:
    """Validate configuration at runtime.

    Raises:
        ValueError: If configuration is invalid.
    """
    # Validate port
    if not (1 <= self.port <= 65535):
        raise ValueError(f"Invalid port: {self.port}")

    # Ensure document_root is Path (should be set by __post_init__)
    assert isinstance(self.document_root, Path)

    # Validate document root
    if not self.document_root.exists():
        raise ValueError(f"Document root does not exist: {self.document_root}")
    if not self.document_root.is_dir():
        raise ValueError(f"Document root is not a directory: {self.document_root}")

    # Validate upload config
    if self.enable_upload:
        if self.upload_dir is None:
            raise ValueError("upload_dir is required when enable_upload is True")

get_rate_limit_config

get_rate_limit_config()

Get rate limit configuration if enabled.

Returns:

Type	Description
RateLimitConfig | None	RateLimitConfig if enabled, None otherwise.

Source code in src/teyaotlani/server/config.py

def get_rate_limit_config(self) -> RateLimitConfig | None:
    """Get rate limit configuration if enabled.

    Returns:
        RateLimitConfig if enabled, None otherwise.
    """
    if not self.enable_rate_limiting:
        return None

    return RateLimitConfig(
        capacity=self.rate_limit_capacity,
        refill_rate=self.rate_limit_refill_rate,
        retry_after=self.rate_limit_retry_after,
    )

get_access_control_config

get_access_control_config()

Get access control configuration if enabled.

Returns:

Type	Description
AccessControlConfig | None	AccessControlConfig if enabled, None otherwise.

Source code in src/teyaotlani/server/config.py

def get_access_control_config(self) -> AccessControlConfig | None:
    """Get access control configuration if enabled.

    Returns:
        AccessControlConfig if enabled, None otherwise.
    """
    if not self.enable_access_control:
        return None

    return AccessControlConfig(
        allow_list=self.access_control_allow_list,
        deny_list=self.access_control_deny_list,
        default_allow=self.access_control_default_allow,
    )

from_toml classmethod

from_toml(path)

Load configuration from a TOML file.

Parameters:

Name	Type	Description	Default
path	Path	Path to the TOML configuration file.	required

Returns:

Type	Description
ServerConfig	A ServerConfig instance.

Raises:

Type	Description
FileNotFoundError	If the config file doesn't exist.
ValueError	If the config file is invalid.

Source code in src/teyaotlani/server/config.py

@classmethod
def from_toml(cls, path: Path) -> "ServerConfig":
    """Load configuration from a TOML file.

    Args:
        path: Path to the TOML configuration file.

    Returns:
        A ServerConfig instance.

    Raises:
        FileNotFoundError: If the config file doesn't exist.
        ValueError: If the config file is invalid.
    """
    if not path.exists():
        raise FileNotFoundError(f"Config file not found: {path}")

    with open(path, "rb") as f:
        data = tomllib.load(f)

    return cls._from_dict(data)

Server Functions

run_server

teyaotlani.run_server

run_server(
    document_root,
    host="localhost",
    port=DEFAULT_PORT,
    **kwargs,
)

Convenience function to run a server synchronously.

Parameters:

Name	Type	Description	Default
document_root	Path | str	Path to serve files from.	required
host	str	Server host address.	'localhost'
port	int	Server port.	DEFAULT_PORT
**kwargs	object	Additional arguments passed to start_server.	{}

Source code in src/teyaotlani/server/server.py

def run_server(
    document_root: Path | str,
    host: str = "localhost",
    port: int = DEFAULT_PORT,
    **kwargs: object,
) -> None:
    """Convenience function to run a server synchronously.

    Args:
        document_root: Path to serve files from.
        host: Server host address.
        port: Server port.
        **kwargs: Additional arguments passed to start_server.
    """
    asyncio.run(
        start_server(
            host=host,
            port=port,
            document_root=document_root,
            **kwargs,  # type: ignore[arg-type]
        )
    )

start_server

teyaotlani.start_server async

start_server(
    config=None,
    host=None,
    port=None,
    document_root=None,
    enable_directory_listing=False,
    log_level="INFO",
    json_logs=False,
    hash_ips=True,
)

Start a Spartan protocol server.

Parameters:

Name	Type	Description	Default
config	ServerConfig | None	Server configuration. If None, uses other arguments.	None
host	str | None	Server host address (default: localhost).	None
port	int | None	Server port (default: 300).	None
document_root	Path | str | None	Path to serve files from.	None
enable_directory_listing	bool	Whether to enable directory listings.	False
log_level	str	Logging level.	'INFO'
json_logs	bool	Whether to output logs as JSON.	False
hash_ips	bool	Whether to hash client IPs in logs.	True

Raises:

Type	Description
ValueError	If configuration is invalid.

Source code in src/teyaotlani/server/server.py

async def start_server(
    config: ServerConfig | None = None,
    host: str | None = None,
    port: int | None = None,
    document_root: Path | str | None = None,
    enable_directory_listing: bool = False,
    log_level: str = "INFO",
    json_logs: bool = False,
    hash_ips: bool = True,
) -> None:
    """Start a Spartan protocol server.

    Args:
        config: Server configuration. If None, uses other arguments.
        host: Server host address (default: localhost).
        port: Server port (default: 300).
        document_root: Path to serve files from.
        enable_directory_listing: Whether to enable directory listings.
        log_level: Logging level.
        json_logs: Whether to output logs as JSON.
        hash_ips: Whether to hash client IPs in logs.

    Raises:
        ValueError: If configuration is invalid.
    """
    # Build config from arguments if not provided
    if config is None:
        if document_root is None:
            raise ValueError("document_root is required")

        config = ServerConfig(
            host=host or "localhost",
            port=port or DEFAULT_PORT,
            document_root=document_root,
            enable_directory_listing=enable_directory_listing,
            log_level=log_level,
            json_logs=json_logs,
            hash_client_ips=hash_ips,
        )

    # Validate configuration
    config.validate()

    # Configure logging
    configure_logging(
        level=config.log_level,
        json_format=config.json_logs,
        hash_ips=config.hash_client_ips,
    )

    # Create handlers
    static_handler = StaticFileHandler(
        document_root=config.document_root,
        default_indices=config.index_files,
        enable_directory_listing=config.enable_directory_listing,
        max_file_size=config.max_file_size,
    )

    upload_handler = None
    if config.enable_upload and config.upload_dir:
        upload_handler = UploadHandler(
            upload_dir=config.upload_dir,
            max_size=config.max_upload_size,
            enable_delete=config.enable_delete,
        )

    combined_handler = CombinedHandler(static_handler, upload_handler)

    # Create middleware chain
    middleware = MiddlewareChain()

    rate_limiter = None
    rate_limit_config = config.get_rate_limit_config()
    if rate_limit_config:
        rate_limiter = RateLimiter(rate_limit_config)
        rate_limiter.start()
        middleware.add(rate_limiter)

    access_control_config = config.get_access_control_config()
    if access_control_config:
        access_control = AccessControl(access_control_config)
        middleware.add(access_control)

    # Get event loop
    loop = asyncio.get_running_loop()

    # Create server
    server = await loop.create_server(
        lambda: SpartanServerProtocol(
            request_handler=combined_handler.handle,
            middleware=middleware if middleware.middlewares else None,
        ),
        host=config.host,
        port=config.port,
    )

    # Log startup
    logger.info(
        "server_started",
        host=config.host,
        port=config.port,
        document_root=str(config.document_root),
        directory_listing=config.enable_directory_listing,
        upload_enabled=config.enable_upload,
        rate_limiting=config.enable_rate_limiting,
        access_control=config.enable_access_control,
    )

    print(f"Spartan server running on spartan://{config.host}:{config.port}/")
    print(f"Serving files from: {config.document_root}")
    print("Press Ctrl+C to stop")

    # Handle shutdown
    shutdown_event = asyncio.Event()

    def signal_handler() -> None:
        shutdown_event.set()

    # Register signal handlers
    for sig in (signal.SIGINT, signal.SIGTERM):
        loop.add_signal_handler(sig, signal_handler)

    try:
        async with server:
            await shutdown_event.wait()
    finally:
        # Cleanup
        if rate_limiter:
            rate_limiter.stop()

        logger.info("server_stopped")
        print("\nServer stopped.")