Client API Reference

The client module provides functionality for making Spartan protocol requests.

Quick example

import asyncio
from teyaotlani import SpartanClient, get

# Simple one-off request
response = asyncio.run(get("spartan://example.com/"))

# Multiple requests with a client
async def main():
    async with SpartanClient() as client:
        response = await client.get("spartan://example.com/")
        print(response.body)

asyncio.run(main())

SpartanClient

teyaotlani.SpartanClient

SpartanClient(timeout=30.0)

High-level Spartan client with async/await API.

This class provides a simple, high-level interface for making Spartan requests. Handles connection management and timeouts.

Spartan is plaintext by default (no TLS, no TOFU).

Examples:

>>> # Basic usage
>>> async with SpartanClient() as client:
...     response = await client.get("spartan://example.com/")
...     print(response.body)

>>> # With custom timeout
>>> client = SpartanClient(timeout=60)
>>> response = await client.get("spartan://example.com/")

>>> # Upload content
>>> response = await client.upload(
...     "spartan://example.com/file.txt",
...     b"Hello, world!"
... )

Initialize the Spartan client.

Parameters:

Name	Type	Description	Default
timeout	float	Request timeout in seconds.	30.0

Source code in src/teyaotlani/client/session.py

def __init__(self, timeout: float = 30.0) -> None:
    """Initialize the Spartan client.

    Args:
        timeout: Request timeout in seconds.
    """
    self.timeout = timeout

Functions

__aenter__ async

__aenter__()

Async context manager entry.

Source code in src/teyaotlani/client/session.py

async def __aenter__(self) -> "SpartanClient":
    """Async context manager entry."""
    return self

__aexit__ async

__aexit__(exc_type, exc_val, exc_tb)

Async context manager exit.

Source code in src/teyaotlani/client/session.py

async def __aexit__(
    self,
    exc_type: type[BaseException] | None,
    exc_val: BaseException | None,
    exc_tb: TracebackType | None,
) -> None:
    """Async context manager exit."""
    pass

get async

get(url)

Get a Spartan resource.

Parameters:

Name	Type	Description	Default
url	str	The Spartan URL to get.	required

Returns:

Type	Description
SpartanResponse	A SpartanResponse with status, meta, and optional body.

Raises:

Type	Description
ValueError	If the URL is invalid.
TimeoutError	If the request times out.
ConnectionError	If the connection fails.

Examples:

>>> response = await client.get("spartan://example.com/")
>>> if response.is_success():
...     print(response.body)

Source code in src/teyaotlani/client/session.py

async def get(self, url: str) -> SpartanResponse:
    """Get a Spartan resource.

    Args:
        url: The Spartan URL to get.

    Returns:
        A SpartanResponse with status, meta, and optional body.

    Raises:
        ValueError: If the URL is invalid.
        TimeoutError: If the request times out.
        ConnectionError: If the connection fails.

    Examples:
        >>> response = await client.get("spartan://example.com/")
        >>> if response.is_success():
        ...     print(response.body)
    """
    validate_url(url)
    return await self._request(url, b"")

upload async

upload(url, content)

Upload content to a Spartan server.

Spartan integrates upload into the core protocol (content-length > 0 in the request).

Parameters:

Name	Type	Description	Default
url	str	The target URL.	required
content	bytes | str	The content to upload.	required

Returns:

Type	Description
SpartanResponse	A SpartanResponse indicating success or failure.

Raises:

Type	Description
ValueError	If the URL is invalid.
TimeoutError	If the request times out.
ConnectionError	If the connection fails.

Examples:

>>> response = await client.upload(
...     "spartan://example.com/file.txt",
...     "Hello, world!"
... )

Source code in src/teyaotlani/client/session.py

async def upload(
    self,
    url: str,
    content: bytes | str,
) -> SpartanResponse:
    """Upload content to a Spartan server.

    Spartan integrates upload into the core protocol
    (content-length > 0 in the request).

    Args:
        url: The target URL.
        content: The content to upload.

    Returns:
        A SpartanResponse indicating success or failure.

    Raises:
        ValueError: If the URL is invalid.
        TimeoutError: If the request times out.
        ConnectionError: If the connection fails.

    Examples:
        >>> response = await client.upload(
        ...     "spartan://example.com/file.txt",
        ...     "Hello, world!"
        ... )
    """
    validate_url(url)

    # Convert content to bytes
    if isinstance(content, str):
        content_bytes = content.encode("utf-8")
    else:
        content_bytes = content

    return await self._request(url, content_bytes)

query async

query(url, query)

Send a query to a Spartan server.

In Spartan, queries are sent as the data block (like uploads).

Parameters:

Name	Type	Description	Default
url	str	The target URL.	required
query	str	The query string.	required

Returns:

Type	Description
SpartanResponse	A SpartanResponse.

Examples:

>>> response = await client.query(
...     "spartan://example.com/search",
...     "hello world"
... )

Source code in src/teyaotlani/client/session.py

async def query(
    self,
    url: str,
    query: str,
) -> SpartanResponse:
    """Send a query to a Spartan server.

    In Spartan, queries are sent as the data block (like uploads).

    Args:
        url: The target URL.
        query: The query string.

    Returns:
        A SpartanResponse.

    Examples:
        >>> response = await client.query(
        ...     "spartan://example.com/search",
        ...     "hello world"
        ... )
    """
    return await self.upload(url, query)

Convenience Functions

get

teyaotlani.get async

get(url, timeout=30.0)

Convenience function to get a Spartan URL.

Parameters:

Name	Type	Description	Default
url	str	The Spartan URL to get.	required
timeout	float	Request timeout in seconds.	30.0

Returns:

Type	Description
SpartanResponse	A SpartanResponse.

Examples:

>>> response = await teyaotlani.get("spartan://example.com/")

Source code in src/teyaotlani/client/session.py

async def get(url: str, timeout: float = 30.0) -> SpartanResponse:
    """Convenience function to get a Spartan URL.

    Args:
        url: The Spartan URL to get.
        timeout: Request timeout in seconds.

    Returns:
        A SpartanResponse.

    Examples:
        >>> response = await teyaotlani.get("spartan://example.com/")
    """
    async with SpartanClient(timeout=timeout) as client:
        return await client.get(url)

upload

teyaotlani.upload async

upload(url, content, timeout=30.0)

Convenience function to upload content.

Parameters:

Name	Type	Description	Default
url	str	The target URL.	required
content	bytes | str	The content to upload.	required
timeout	float	Request timeout in seconds.	30.0

Returns:

Type	Description
SpartanResponse	A SpartanResponse.

Source code in src/teyaotlani/client/session.py

async def upload(
    url: str,
    content: bytes | str,
    timeout: float = 30.0,
) -> SpartanResponse:
    """Convenience function to upload content.

    Args:
        url: The target URL.
        content: The content to upload.
        timeout: Request timeout in seconds.

    Returns:
        A SpartanResponse.
    """
    async with SpartanClient(timeout=timeout) as client:
        return await client.upload(url, content)