Server API

API reference for the Mototli Gopher server.

GopherServer

GopherServer

GopherServer(config: ServerConfig)

High-level Gopher server.

This class provides a simple interface for creating and running a Gopher server with static file serving, CGI support, and Gopher+ extensions.

Attributes:

Name	Type	Description
config		Server configuration.
router		Request router.
server	Server | None	The asyncio server object when running.

Examples:

>>> config = ServerConfig(
...     host="localhost",
...     port=70,
...     document_root=Path("/var/gopher"),
... )
>>> server = GopherServer(config)
>>> await server.start()

Initialize the Gopher server.

Parameters:

Name	Type	Description	Default
config	ServerConfig	Server configuration.	required

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

def __init__(self, config: ServerConfig) -> None:
    """Initialize the Gopher server.

    Args:
        config: Server configuration.
    """
    self.config = config
    self.router = Router()
    self.server: asyncio.Server | None = None
    self._setup_routes()

serve_forever async

serve_forever() -> None

Run the server forever.

This method blocks until the server is stopped.

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

async def serve_forever(self) -> None:
    """Run the server forever.

    This method blocks until the server is stopped.
    """
    if not self.server:
        await self.start()

    if self.server:
        async with self.server:
            await self.server.serve_forever()

start async

start() -> None

Start the server.

Creates the asyncio server and starts listening for connections.

Raises:

Type	Description
OSError	If unable to bind to the specified host/port.

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

async def start(self) -> None:
    """Start the server.

    Creates the asyncio server and starts listening for connections.

    Raises:
        OSError: If unable to bind to the specified host/port.
    """
    # Validate configuration
    self.config.validate()

    # Get event loop
    loop = asyncio.get_running_loop()

    # Create server
    self.server = await loop.create_server(
        lambda: GopherServerProtocol(
            request_handler=self.router.route,
            request_timeout=self.config.request_timeout,
        ),
        self.config.host,
        self.config.port,
    )

    print(f"Gopher server started on {self.config.host}:{self.config.port}")
    print(f"Document root: {self.config.document_root}")
    if self.config.gopher_plus:
        print("Gopher+ enabled")

stop async

stop() -> None

Stop the server gracefully.

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

async def stop(self) -> None:
    """Stop the server gracefully."""
    if self.server:
        self.server.close()
        await self.server.wait_closed()
        self.server = None
        print("Gopher server stopped")

ServerConfig

ServerConfig dataclass

ServerConfig(
    host: str = "localhost",
    port: int = DEFAULT_PORT,
    document_root: Path = (lambda: Path("."))(),
    hostname: str | None = None,
    enable_directory_listing: bool = True,
    default_indices: list[str] = (
        lambda: ["index.gph", "gophermap", "index.txt"]
    )(),
    cgi_extensions: list[str] = (
        lambda: [".cgi", ".sh", ".py", ".pl"]
    )(),
    cgi_directories: list[str] = (lambda: ["cgi-bin"])(),
    max_file_size: int = 100 * 1024 * 1024,
    request_timeout: float = REQUEST_TIMEOUT,
    cgi_timeout: float = 30.0,
    gopher_plus: bool = True,
    admin_name: str | None = None,
    admin_email: str | None = None,
)

Configuration for the Gopher server.

Attributes:

Name	Type	Description
host	str	The host address to bind to.
port	int	The port to listen on (default 70).
document_root	Path	Path to the directory containing files to serve.
hostname	str | None	Public hostname for generating menu items. If None, uses host.
enable_directory_listing	bool	Whether to generate directory listings.
default_indices	list[str]	List of index filenames to try for directory requests.
cgi_extensions	list[str]	File extensions that indicate CGI scripts.
cgi_directories	list[str]	Directory names that contain CGI scripts.
max_file_size	int	Maximum file size to serve in bytes.
request_timeout	float	Timeout for receiving requests in seconds.
cgi_timeout	float	Timeout for CGI script execution in seconds.
gopher_plus	bool	Whether Gopher+ is enabled.
admin_name	str | None	Administrator name for Gopher+ ADMIN block.
admin_email	str | None	Administrator email for Gopher+ ADMIN block.

Examples:

>>> config = ServerConfig(
...     host="localhost",
...     port=70,
...     document_root=Path("/var/gopher"),
... )

public_hostname property

public_hostname: str

Get the public hostname for menu items.

public_port property

public_port: int

Get the public port for menu items.

__post_init__

__post_init__() -> None

Validate and normalize configuration after initialization.

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

def __post_init__(self) -> None:
    """Validate and normalize configuration after initialization."""
    # Ensure document_root is a Path
    if isinstance(self.document_root, str):
        self.document_root = Path(self.document_root)

    # Resolve to absolute path
    self.document_root = self.document_root.resolve()

    # Use host as hostname if not specified
    if self.hostname is None:
        self.hostname = self.host

from_toml classmethod

from_toml(path: Path | str) -> ServerConfig

Load configuration from a TOML file.

Parameters:

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

Returns:

Type	Description
ServerConfig	A ServerConfig instance with values from the file.

Raises:

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

Examples:

>>> config = ServerConfig.from_toml("config.toml")

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

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

    Args:
        path: Path to the TOML configuration file.

    Returns:
        A ServerConfig instance with values from the file.

    Raises:
        FileNotFoundError: If the config file doesn't exist.
        ValueError: If the config file is invalid.

    Examples:
        >>> config = ServerConfig.from_toml("config.toml")
    """
    import sys

    if sys.version_info >= (3, 11):
        import tomllib
    else:
        import tomli as tomllib

    path = Path(path)
    if not path.exists():
        raise FileNotFoundError(f"Config file not found: {path}")

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

    return cls._from_dict(data)

to_toml

to_toml() -> str

Serialize the configuration to a TOML string.

Returns:

Type	Description
str	The configuration as a TOML-formatted string.

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

def to_toml(self) -> str:
    """Serialize the configuration to a TOML string.

    Returns:
        The configuration as a TOML-formatted string.
    """
    lines = [
        "# Mototli Gopher Server Configuration",
        "",
        "[server]",
        f'host = "{self.host}"',
        f"port = {self.port}",
        f'document_root = "{self.document_root}"',
    ]

    if self.hostname:
        lines.append(f'hostname = "{self.hostname}"')

    lines.extend(
        [
            "",
            "[handlers]",
            f"enable_directory_listing = {str(self.enable_directory_listing).lower()}",  # noqa: E501
            f"default_indices = {self.default_indices!r}",
            f"cgi_extensions = {self.cgi_extensions!r}",
            f"cgi_directories = {self.cgi_directories!r}",
            "",
            "[gopher_plus]",
            f"enabled = {str(self.gopher_plus).lower()}",
        ]
    )

    if self.admin_name:
        lines.append(f'admin_name = "{self.admin_name}"')
    if self.admin_email:
        lines.append(f'admin_email = "{self.admin_email}"')

    lines.extend(
        [
            "",
            "[limits]",
            f"max_file_size = {self.max_file_size}",
            f"request_timeout = {self.request_timeout}",
            f"cgi_timeout = {self.cgi_timeout}",
        ]
    )

    return "\n".join(lines) + "\n"

validate

validate() -> None

Validate the configuration.

Raises:

Type	Description
ValueError	If the configuration is invalid.

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

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

    Raises:
        ValueError: If the configuration is invalid.
    """
    # Check document root exists
    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}")

    # Check port is valid
    if not 1 <= self.port <= 65535:
        raise ValueError(f"Port must be between 1 and 65535, got: {self.port}")

    # Check timeouts are positive
    if self.request_timeout <= 0:
        raise ValueError(
            f"Request timeout must be positive, got: {self.request_timeout}"
        )
    if self.cgi_timeout <= 0:
        raise ValueError(f"CGI timeout must be positive, got: {self.cgi_timeout}")

    # Check file size limit is positive
    if self.max_file_size <= 0:
        raise ValueError(f"Max file size must be positive, got: {self.max_file_size}")

Router

Router

Router()

Routes incoming requests to appropriate handlers.

The Router supports two types of route matching: - Exact: Selector must match exactly - Prefix: Selector must start with the pattern

Routes are matched in the order they were registered.

Examples:

>>> router = Router()
>>> router.add_route("/", index_handler)
>>> router.add_route("/files/", file_handler, route_type=RouteType.PREFIX)

Initialize an empty router.

Source code in src/mototli/server/router.py

def __init__(self) -> None:
    """Initialize an empty router."""
    self.routes: list[Route] = []
    self.default_handler: Callable[[GopherRequest], GopherResponse] | None = None

add_route

add_route(
    pattern: str,
    handler: Callable[[GopherRequest], GopherResponse],
    route_type: RouteType = RouteType.EXACT,
) -> None

Register a new route.

Parameters:

Name	Type	Description	Default
pattern	str	The selector pattern to match.	required
handler	Callable[[GopherRequest], GopherResponse]	Callable that takes a GopherRequest and returns a GopherResponse.	required
route_type	RouteType	Type of matching to perform (default: EXACT).	EXACT

Examples:

>>> router.add_route("/", index_handler)
>>> router.add_route("/cgi-bin/", cgi_handler, RouteType.PREFIX)

Source code in src/mototli/server/router.py

def add_route(
    self,
    pattern: str,
    handler: Callable[[GopherRequest], GopherResponse],
    route_type: RouteType = RouteType.EXACT,
) -> None:
    """Register a new route.

    Args:
        pattern: The selector pattern to match.
        handler: Callable that takes a GopherRequest and returns a GopherResponse.
        route_type: Type of matching to perform (default: EXACT).

    Examples:
        >>> router.add_route("/", index_handler)
        >>> router.add_route("/cgi-bin/", cgi_handler, RouteType.PREFIX)
    """
    route = Route(
        pattern=pattern,
        handler=handler,
        route_type=route_type,
    )
    self.routes.append(route)

route

route(request: GopherRequest) -> GopherResponse

Route a request to the appropriate handler.

Routes are matched in the order they were registered. If no route matches, the default handler is called (if set), otherwise an error response is returned.

Parameters:

Name	Type	Description	Default
request	GopherRequest	The incoming request to route.	required

Returns:

Type	Description
GopherResponse	The response from the matched handler.

Source code in src/mototli/server/router.py

def route(self, request: GopherRequest) -> GopherResponse:
    """Route a request to the appropriate handler.

    Routes are matched in the order they were registered.
    If no route matches, the default handler is called (if set),
    otherwise an error response is returned.

    Args:
        request: The incoming request to route.

    Returns:
        The response from the matched handler.
    """
    selector = request.selector

    # Try to match against registered routes
    for route in self.routes:
        if self._matches(selector, route):
            return route.handler(request)

    # No match found, use default handler or return error
    if self.default_handler:
        return self.default_handler(request)

    # No default handler, return generic error
    return GopherResponse(
        items=[create_error_item("Not found")],
        is_directory=True,
    )

set_default_handler

set_default_handler(
    handler: Callable[[GopherRequest], GopherResponse],
) -> None

Set the default handler for unmatched routes.

Parameters:

Name	Type	Description	Default
handler	Callable[[GopherRequest], GopherResponse]	Callable that handles requests not matching any route. Typically returns an error response.	required

Source code in src/mototli/server/router.py

def set_default_handler(
    self, handler: Callable[[GopherRequest], GopherResponse]
) -> None:
    """Set the default handler for unmatched routes.

    Args:
        handler: Callable that handles requests not matching any route.
            Typically returns an error response.
    """
    self.default_handler = handler

RouteType

RouteType

Bases: Enum

Type of route pattern matching.

EXACT class-attribute instance-attribute

EXACT = auto()

Exact selector match.

PREFIX class-attribute instance-attribute

PREFIX = auto()

Selector prefix match.

Convenience Functions

run_server

run_server async

run_server(config: ServerConfig) -> None

Run a Gopher server with the given configuration.

This is a convenience function that creates a server and runs it until interrupted (Ctrl+C).

Parameters:

Name	Type	Description	Default
config	ServerConfig	Server configuration.	required

Examples:

>>> config = ServerConfig(
...     host="localhost",
...     port=70,
...     document_root=Path("/var/gopher"),
... )
>>> asyncio.run(run_server(config))

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

async def run_server(config: ServerConfig) -> None:
    """Run a Gopher server with the given configuration.

    This is a convenience function that creates a server and runs it
    until interrupted (Ctrl+C).

    Args:
        config: Server configuration.

    Examples:
        >>> config = ServerConfig(
        ...     host="localhost",
        ...     port=70,
        ...     document_root=Path("/var/gopher"),
        ... )
        >>> asyncio.run(run_server(config))
    """
    server = GopherServer(config)

    # Set up signal handlers for graceful shutdown
    loop = asyncio.get_running_loop()

    def signal_handler() -> None:
        print("\nShutting down...")
        asyncio.create_task(server.stop())

    try:
        loop.add_signal_handler(signal.SIGINT, signal_handler)
        loop.add_signal_handler(signal.SIGTERM, signal_handler)
    except NotImplementedError:
        # Windows doesn't support add_signal_handler
        pass

    try:
        await server.serve_forever()
    except asyncio.CancelledError:
        await server.stop()

start_server

start_server async

start_server(
    host: str = "localhost",
    port: int = 70,
    document_root: Path | str = ".",
    hostname: str | None = None,
    enable_directory_listing: bool = True,
    gopher_plus: bool = True,
    admin_name: str | None = None,
    admin_email: str | None = None,
) -> None

Start a Gopher server with the given options.

This is a convenience function that creates a configuration and runs the server.

Parameters:

Name	Type	Description	Default
host	str	Host address to bind to.	'localhost'
port	int	Port to listen on.	70
document_root	Path | str	Path to the document root directory.	'.'
hostname	str | None	Public hostname for menu items. If None, uses host.	None
enable_directory_listing	bool	Whether to generate directory listings.	True
gopher_plus	bool	Whether Gopher+ is enabled.	True
admin_name	str | None	Administrator name for Gopher+ ADMIN block.	None
admin_email	str | None	Administrator email for Gopher+ ADMIN block.	None

Examples:

>>> asyncio.run(start_server(
...     host="0.0.0.0",
...     port=70,
...     document_root="/var/gopher",
...     hostname="gopher.example.com",
... ))

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

async def start_server(
    host: str = "localhost",
    port: int = 70,
    document_root: Path | str = ".",
    hostname: str | None = None,
    enable_directory_listing: bool = True,
    gopher_plus: bool = True,
    admin_name: str | None = None,
    admin_email: str | None = None,
) -> None:
    """Start a Gopher server with the given options.

    This is a convenience function that creates a configuration and runs
    the server.

    Args:
        host: Host address to bind to.
        port: Port to listen on.
        document_root: Path to the document root directory.
        hostname: Public hostname for menu items. If None, uses host.
        enable_directory_listing: Whether to generate directory listings.
        gopher_plus: Whether Gopher+ is enabled.
        admin_name: Administrator name for Gopher+ ADMIN block.
        admin_email: Administrator email for Gopher+ ADMIN block.

    Examples:
        >>> asyncio.run(start_server(
        ...     host="0.0.0.0",
        ...     port=70,
        ...     document_root="/var/gopher",
        ...     hostname="gopher.example.com",
        ... ))
    """
    config = ServerConfig(
        host=host,
        port=port,
        document_root=Path(document_root),
        hostname=hostname,
        enable_directory_listing=enable_directory_listing,
        gopher_plus=gopher_plus,
        admin_name=admin_name,
        admin_email=admin_email,
    )

    await run_server(config)

Handlers

RequestHandler

RequestHandler

Bases: ABC

Abstract base class for request handlers.

All request handlers should inherit from this class and implement the handle() method.

handle abstractmethod

handle(request: GopherRequest) -> GopherResponse

Handle a Gopher request and return a response.

Parameters:

Name	Type	Description	Default
request	GopherRequest	The incoming request to handle.	required

Returns:

Type	Description
GopherResponse	A GopherResponse object.

Source code in src/mototli/server/handler.py

@abstractmethod
def handle(self, request: GopherRequest) -> GopherResponse:
    """Handle a Gopher request and return a response.

    Args:
        request: The incoming request to handle.

    Returns:
        A GopherResponse object.
    """
    pass

StaticFileHandler

StaticFileHandler

StaticFileHandler(
    document_root: Path | str,
    hostname: str,
    port: int = DEFAULT_PORT,
    default_indices: list[str] | None = None,
    enable_directory_listing: bool = True,
    max_file_size: int = 100 * 1024 * 1024,
    gopher_plus: bool = True,
    admin_name: str | None = None,
    admin_email: str | None = None,
)

Bases: RequestHandler

Handler for serving static files from a document root.

This handler serves files from a specified directory with path traversal protection, automatic directory listings, and Gopher+ attribute support.

Attributes:

Name	Type	Description
document_root		Path to the directory containing files to serve.
hostname		Server hostname for generating menu items.
port		Server port for generating menu items.
default_indices		List of index filenames to try for directory requests.
enable_directory_listing		Whether to generate directory listings.
max_file_size		Maximum file size to serve in bytes.
gopher_plus		Whether Gopher+ is enabled.
admin_name		Administrator name for Gopher+ ADMIN block.
admin_email		Administrator email for Gopher+ ADMIN block.

Examples:

>>> handler = StaticFileHandler(
...     Path("/var/gopher"),
...     hostname="gopher.example.com",
... )
>>> request = GopherRequest(selector="/readme.txt")
>>> response = handler.handle(request)

Initialize the static file handler.

Parameters:

Name	Type	Description	Default
document_root	Path | str	Path to the directory containing files to serve.	required
hostname	str	Server hostname for generating menu items.	required
port	int	Server port for generating menu items.	DEFAULT_PORT
default_indices	list[str] | None	List of index filenames to try for directory requests.	None
enable_directory_listing	bool	Whether to generate directory listings.	True
max_file_size	int	Maximum file size to serve in bytes.	100 * 1024 * 1024
gopher_plus	bool	Whether Gopher+ is enabled.	True
admin_name	str | None	Administrator name for Gopher+ ADMIN block.	None
admin_email	str | None	Administrator email for Gopher+ ADMIN block.	None

Source code in src/mototli/server/handler.py

def __init__(
    self,
    document_root: Path | str,
    hostname: str,
    port: int = DEFAULT_PORT,
    default_indices: list[str] | None = None,
    enable_directory_listing: bool = True,
    max_file_size: int = 100 * 1024 * 1024,
    gopher_plus: bool = True,
    admin_name: str | None = None,
    admin_email: str | None = None,
) -> None:
    """Initialize the static file handler.

    Args:
        document_root: Path to the directory containing files to serve.
        hostname: Server hostname for generating menu items.
        port: Server port for generating menu items.
        default_indices: List of index filenames to try for directory requests.
        enable_directory_listing: Whether to generate directory listings.
        max_file_size: Maximum file size to serve in bytes.
        gopher_plus: Whether Gopher+ is enabled.
        admin_name: Administrator name for Gopher+ ADMIN block.
        admin_email: Administrator email for Gopher+ ADMIN block.
    """
    self.document_root = Path(document_root).resolve()
    self.hostname = hostname
    self.port = port
    self.default_indices = default_indices or [
        "index.gph",
        "gophermap",
        "index.txt",
    ]
    self.enable_directory_listing = enable_directory_listing
    self.max_file_size = max_file_size
    self.gopher_plus = gopher_plus
    self.admin_name = admin_name
    self.admin_email = admin_email

    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}")

handle

handle(request: GopherRequest) -> GopherResponse

Handle a request for a static file or directory.

Parameters:

Name	Type	Description	Default
request	GopherRequest	The incoming request.	required

Returns:

Type	Description
GopherResponse	A GopherResponse with the file contents, directory listing,
GopherResponse	or Gopher+ attributes.

Source code in src/mototli/server/handler.py

def handle(self, request: GopherRequest) -> GopherResponse:
    """Handle a request for a static file or directory.

    Args:
        request: The incoming request.

    Returns:
        A GopherResponse with the file contents, directory listing,
        or Gopher+ attributes.
    """
    # Get the requested path
    selector = request.selector.lstrip("/")
    if selector == "":
        selector = "."

    # Construct the full file path
    file_path = (self.document_root / selector).resolve()

    # Path traversal protection
    if not self._is_safe_path(file_path):
        return GopherResponse(
            items=[create_error_item("Not found")],
            is_directory=True,
        )

    # Check if path exists
    if not file_path.exists():
        return GopherResponse(
            items=[create_error_item("Not found")],
            is_directory=True,
        )

    # Handle Gopher+ attribute requests
    if request.request_type == RequestType.ATTRIBUTES:
        return self._handle_attributes_request(request, file_path)

    # Handle Gopher+ directory attribute requests
    if request.request_type == RequestType.DIRECTORY:
        return self._handle_directory_attributes_request(request, file_path)

    # Handle directory
    if file_path.is_dir():
        return self._handle_directory(request, file_path)

    # Handle file
    return self._handle_file(request, file_path)

CGIHandler

CGIHandler

CGIHandler(
    document_root: Path | str,
    hostname: str,
    port: int = DEFAULT_PORT,
    cgi_extensions: list[str] | None = None,
    cgi_directories: list[str] | None = None,
    timeout: float = 30.0,
    gopher_plus: bool = True,
)

Bases: RequestHandler

Handler for executing CGI scripts.

This handler executes CGI scripts and returns their output as Gopher responses. Scripts can be identified by file extension (e.g., .cgi) or by being located in a CGI directory (e.g., cgi-bin/).

The handler sets up standard CGI environment variables plus Gopher-specific variables like SELECTOR and GOPHER_PLUS.

Attributes:

Name	Type	Description
document_root		Path to the document root.
hostname		Server hostname for environment variables.
port		Server port for environment variables.
cgi_extensions		File extensions that indicate CGI scripts.
cgi_directories		Directory names that contain CGI scripts.
timeout		Timeout for CGI script execution in seconds.
gopher_plus		Whether Gopher+ is enabled.

Examples:

>>> handler = CGIHandler(
...     document_root=Path("/var/gopher"),
...     hostname="gopher.example.com",
...     cgi_extensions=[".cgi", ".sh"],
...     cgi_directories=["cgi-bin"],
... )

Initialize the CGI handler.

Parameters:

Name	Type	Description	Default
document_root	Path | str	Path to the document root.	required
hostname	str	Server hostname for environment variables.	required
port	int	Server port for environment variables.	DEFAULT_PORT
cgi_extensions	list[str] | None	File extensions that indicate CGI scripts.	None
cgi_directories	list[str] | None	Directory names that contain CGI scripts.	None
timeout	float	Timeout for CGI script execution in seconds.	30.0
gopher_plus	bool	Whether Gopher+ is enabled.	True

Source code in src/mototli/server/cgi.py

def __init__(
    self,
    document_root: Path | str,
    hostname: str,
    port: int = DEFAULT_PORT,
    cgi_extensions: list[str] | None = None,
    cgi_directories: list[str] | None = None,
    timeout: float = 30.0,
    gopher_plus: bool = True,
) -> None:
    """Initialize the CGI handler.

    Args:
        document_root: Path to the document root.
        hostname: Server hostname for environment variables.
        port: Server port for environment variables.
        cgi_extensions: File extensions that indicate CGI scripts.
        cgi_directories: Directory names that contain CGI scripts.
        timeout: Timeout for CGI script execution in seconds.
        gopher_plus: Whether Gopher+ is enabled.
    """
    self.document_root = Path(document_root).resolve()
    self.hostname = hostname
    self.port = port
    self.cgi_extensions = cgi_extensions or [".cgi", ".sh", ".py", ".pl"]
    self.cgi_directories = cgi_directories or ["cgi-bin"]
    self.timeout = timeout
    self.gopher_plus = gopher_plus

can_handle

can_handle(selector: str) -> bool

Check if this handler can handle a selector.

This is useful for routing decisions.

Parameters:

Name	Type	Description	Default
selector	str	The selector to check.	required

Returns:

Type	Description
bool	True if this handler can handle the selector.

Source code in src/mototli/server/cgi.py

def can_handle(self, selector: str) -> bool:
    """Check if this handler can handle a selector.

    This is useful for routing decisions.

    Args:
        selector: The selector to check.

    Returns:
        True if this handler can handle the selector.
    """
    # Check if in a CGI directory
    selector_parts = selector.strip("/").split("/")
    for part in selector_parts:
        if part in self.cgi_directories:
            return True

    # Check extension
    path = Path(selector)
    if path.suffix.lower() in self.cgi_extensions:
        return True

    return False

execute_cgi_async async

execute_cgi_async(
    script_path: Path, request: GopherRequest
) -> GopherResponse

Execute a CGI script asynchronously.

This method uses asyncio subprocess for non-blocking execution.

Parameters:

Name	Type	Description	Default
script_path	Path	Path to the CGI script.	required
request	GopherRequest	The incoming request.	required

Returns:

Type	Description
GopherResponse	A GopherResponse with the script output.

Source code in src/mototli/server/cgi.py

async def execute_cgi_async(
    self, script_path: Path, request: GopherRequest
) -> GopherResponse:
    """Execute a CGI script asynchronously.

    This method uses asyncio subprocess for non-blocking execution.

    Args:
        script_path: Path to the CGI script.
        request: The incoming request.

    Returns:
        A GopherResponse with the script output.
    """
    # Build environment variables
    env = self._build_cgi_env(script_path, request)

    try:
        # Execute the script
        proc = await asyncio.create_subprocess_exec(
            str(script_path),
            stdout=asyncio.subprocess.PIPE,
            stderr=asyncio.subprocess.PIPE,
            env=env,
            cwd=script_path.parent,
        )

        try:
            stdout, stderr = await asyncio.wait_for(
                proc.communicate(),
                timeout=self.timeout,
            )
        except asyncio.TimeoutError:
            proc.kill()
            await proc.wait()
            return GopherResponse(
                items=[create_error_item("CGI script timeout")],
                is_directory=True,
            )

        # Check for errors
        if proc.returncode != 0:
            error_msg = stderr.decode("utf-8", errors="replace").strip()
            if not error_msg:
                error_msg = f"Script exited with code {proc.returncode}"
            return GopherResponse(
                items=[create_error_item(f"CGI error: {error_msg}")],
                is_directory=True,
            )

        # Parse the output
        return self._parse_cgi_output(stdout)

    except PermissionError:
        return GopherResponse(
            items=[create_error_item("Permission denied")],
            is_directory=True,
        )

    except OSError as e:
        return GopherResponse(
            items=[create_error_item(f"Cannot execute script: {e}")],
            is_directory=True,
        )

handle

handle(request: GopherRequest) -> GopherResponse

Handle a request by executing the CGI script.

This method runs synchronously but calls an async helper to execute the CGI script.

Parameters:

Name	Type	Description	Default
request	GopherRequest	The incoming request.	required

Returns:

Type	Description
GopherResponse	A GopherResponse with the CGI script output.

Source code in src/mototli/server/cgi.py

def handle(self, request: GopherRequest) -> GopherResponse:
    """Handle a request by executing the CGI script.

    This method runs synchronously but calls an async helper
    to execute the CGI script.

    Args:
        request: The incoming request.

    Returns:
        A GopherResponse with the CGI script output.
    """
    # Get the requested path
    selector = request.selector.lstrip("/")
    if not selector:
        return GopherResponse(
            items=[create_error_item("Invalid CGI request")],
            is_directory=True,
        )

    # Construct the script path
    script_path = (self.document_root / selector).resolve()

    # Path traversal protection
    if not self._is_safe_path(script_path):
        return GopherResponse(
            items=[create_error_item("Not found")],
            is_directory=True,
        )

    # Check if file exists and is executable
    if not script_path.exists():
        return GopherResponse(
            items=[create_error_item("Not found")],
            is_directory=True,
        )

    if not script_path.is_file():
        return GopherResponse(
            items=[create_error_item("Not a file")],
            is_directory=True,
        )

    if not os.access(script_path, os.X_OK):
        return GopherResponse(
            items=[create_error_item("Script not executable")],
            is_directory=True,
        )

    # Verify this is a CGI script
    if not self._is_cgi_script(script_path, selector):
        return GopherResponse(
            items=[create_error_item("Not a CGI script")],
            is_directory=True,
        )

    # Execute the script
    try:
        return self._execute_cgi_sync(script_path, request)
    except Exception as e:
        return GopherResponse(
            items=[create_error_item(f"CGI error: {e}")],
            is_directory=True,
        )

ErrorHandler

ErrorHandler

ErrorHandler(message: str = 'Error')

Bases: RequestHandler

Handler that returns error responses.

Useful for handling 404 Not Found and other error cases.

Examples:

>>> handler = ErrorHandler("Not found")
>>> response = handler.handle(request)

Initialize the error handler.

Parameters:

Name	Type	Description	Default
message	str	The error message to return.	'Error'

Source code in src/mototli/server/handler.py

def __init__(self, message: str = "Error") -> None:
    """Initialize the error handler.

    Args:
        message: The error message to return.
    """
    self.message = message

handle

handle(request: GopherRequest) -> GopherResponse

Return an error response.

Parameters:

Name	Type	Description	Default
request	GopherRequest	The incoming request (ignored).	required

Returns:

Type	Description
GopherResponse	A GopherResponse with an error item.

Source code in src/mototli/server/handler.py

def handle(self, request: GopherRequest) -> GopherResponse:
    """Return an error response.

    Args:
        request: The incoming request (ignored).

    Returns:
        A GopherResponse with an error item.
    """
    return GopherResponse(
        items=[create_error_item(self.message)],
        is_directory=True,
    )

Usage Examples

Quick Start

import asyncio
from mototli.server import run_server

# Serve current directory on port 7070
asyncio.run(run_server(document_root=".", port=7070))

With Configuration

import asyncio
from mototli.server import GopherServer, ServerConfig
from pathlib import Path

config = ServerConfig(
    host="0.0.0.0",
    port=7070,
    hostname="gopher.example.com",
    document_root=Path("/var/gopher"),
    gopher_plus=True,
    admin_name="Admin",
    admin_email="admin@example.com"
)

async def main():
    server = GopherServer(config)
    await server.start()
    try:
        await asyncio.Event().wait()  # Run forever
    finally:
        await server.stop()

asyncio.run(main())

From TOML File

import asyncio
from mototli.server import GopherServer, ServerConfig

config = ServerConfig.from_toml("server.toml")

async def main():
    server = GopherServer(config)
    await server.start()
    await asyncio.Event().wait()

asyncio.run(main())

Custom Routing

from mototli.server import Router, RouteType

router = Router()

# Exact match
router.add_route("/about", about_handler, RouteType.EXACT)

# Prefix match
router.add_route("/files/", file_handler, RouteType.PREFIX)

Custom Handler

from mototli.server import RequestHandler
from mototli.protocol import GopherRequest, GopherResponse

class CustomHandler(RequestHandler):
    async def handle(self, request: GopherRequest) -> GopherResponse:
        # Generate custom response
        content = f"You requested: {request.selector}"
        return GopherResponse(content=content.encode())

See Also

• Your First Gopherhole - Tutorial
• Configure Server - Configuration guide
• Configuration Reference - TOML options