Skip to content

Client API

API reference for the Mototli Gopher client.

GopherClient

GopherClient 

GopherClient(timeout: float = REQUEST_TIMEOUT)

High-level Gopher client with async/await API.

This class provides a simple, high-level interface for getting Gopher resources. It handles connection management, timeouts, and response parsing.

Examples:

>>> # Basic usage
>>> async with GopherClient() as client:
...     response = await client.get("gopher.floodgap.com")
...     for item in response.items:
...         print(item)

>>> # Get a specific resource
>>> response = await client.get(
...     "gopher.floodgap.com",
...     selector="/overbite",
... )

>>> # Get with search query
>>> response = await client.get(
...     "gopher.floodgap.com",
...     selector="/v2/vs",
...     search_query="python",
... )

Initialize the Gopher client.

Parameters:

Name Type Description Default
timeout float

Request timeout in seconds. Default is 30 seconds.

 REQUEST_TIMEOUT
Source code in src/mototli/client/session.py 
44
45
46
47
48
49
50
51
52
53
def __init__(
    self,
    timeout: float = REQUEST_TIMEOUT,
):
    """Initialize the Gopher client.

    Args:
        timeout: Request timeout in seconds. Default is 30 seconds.
    """
    self.timeout = timeout

get async 

get(
    host: str,
    selector: str = "",
    port: int = DEFAULT_PORT,
    search_query: str | None = None,
    item_type: ItemType = ItemType.DIRECTORY,
) -> GopherResponse

Get a Gopher resource.

Parameters:

Name Type Description Default
host str

The Gopher server hostname.

 required
selector str

The resource selector (path). Default is empty (root).

 ''
port int

The server port. Default is 70.

 DEFAULT_PORT
search_query str | None

Optional search query for type 7 (search) items.

 None
item_type ItemType

Expected item type. Used to determine if response is a directory listing. Default is DIRECTORY.

 DIRECTORY

Returns:

Type Description
GopherResponse

A GopherResponse object with items (for directories) or raw_body.

Raises:

Type Description
TimeoutError

If the request times out.
ConnectionError

If the connection fails.

Examples:

>>> response = await client.get("gopher.floodgap.com")
>>> for item in response.items:
...     print(f"[{item.item_type.value}] {item.display_text}")
Source code in src/mototli/client/session.py 
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
async def get(
    self,
    host: str,
    selector: str = "",
    port: int = DEFAULT_PORT,
    search_query: str | None = None,
    item_type: ItemType = ItemType.DIRECTORY,
) -> GopherResponse:
    """Get a Gopher resource.

    Args:
        host: The Gopher server hostname.
        selector: The resource selector (path). Default is empty (root).
        port: The server port. Default is 70.
        search_query: Optional search query for type 7 (search) items.
        item_type: Expected item type. Used to determine if response is a
            directory listing. Default is DIRECTORY.

    Returns:
        A GopherResponse object with items (for directories) or raw_body.

    Raises:
        asyncio.TimeoutError: If the request times out.
        ConnectionError: If the connection fails.

    Examples:
        >>> response = await client.get("gopher.floodgap.com")
        >>> for item in response.items:
        ...     print(f"[{item.item_type.value}] {item.display_text}")
    """
    request = GopherRequest(
        selector=selector,
        search_query=search_query,
    )

    is_directory = item_type.is_directory or item_type == ItemType.SEARCH

    return await self._send_request(
        host=host,
        port=port,
        request=request,
        is_directory=is_directory,
    )

get_text async 

get_text(
    host: str, selector: str, port: int = DEFAULT_PORT
) -> str

Get a text resource and return its content as a string.

Parameters:

Name Type Description Default
host str

The Gopher server hostname.

 required
selector str

The resource selector (path).

 required
port int

The server port. Default is 70.

 DEFAULT_PORT

Returns:

Type Description
str

The text content as a string.

Raises:

Type Description
TimeoutError

If the request times out.
ConnectionError

If the connection fails.
ValueError

If the response is not text.

Examples:

>>> text = await client.get_text(
...     "gopher.floodgap.com",
...     "/gopher/welcome",
... )
>>> print(text)
Source code in src/mototli/client/session.py 
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
async def get_text(
    self,
    host: str,
    selector: str,
    port: int = DEFAULT_PORT,
) -> str:
    """Get a text resource and return its content as a string.

    Args:
        host: The Gopher server hostname.
        selector: The resource selector (path).
        port: The server port. Default is 70.

    Returns:
        The text content as a string.

    Raises:
        asyncio.TimeoutError: If the request times out.
        ConnectionError: If the connection fails.
        ValueError: If the response is not text.

    Examples:
        >>> text = await client.get_text(
        ...     "gopher.floodgap.com",
        ...     "/gopher/welcome",
        ... )
        >>> print(text)
    """
    response = await self.get(
        host=host,
        selector=selector,
        port=port,
        item_type=ItemType.TEXT,
    )

    if response.text is None:
        raise ValueError("Response is not text content")

    return response.text

get_binary async 

get_binary(
    host: str, selector: str, port: int = DEFAULT_PORT
) -> bytes

Get a binary resource.

Parameters:

Name Type Description Default
host str

The Gopher server hostname.

 required
selector str

The resource selector (path).

 required
port int

The server port. Default is 70.

 DEFAULT_PORT

Returns:

Type Description
bytes

The binary content as bytes.

Raises:

Type Description
TimeoutError

If the request times out.
ConnectionError

If the connection fails.

Examples:

>>> data = await client.get_binary(
...     "gopher.floodgap.com",
...     "/some/image.gif",
... )
>>> with open("image.gif", "wb") as f:
...     f.write(data)
Source code in src/mototli/client/session.py 
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
async def get_binary(
    self,
    host: str,
    selector: str,
    port: int = DEFAULT_PORT,
) -> bytes:
    """Get a binary resource.

    Args:
        host: The Gopher server hostname.
        selector: The resource selector (path).
        port: The server port. Default is 70.

    Returns:
        The binary content as bytes.

    Raises:
        asyncio.TimeoutError: If the request times out.
        ConnectionError: If the connection fails.

    Examples:
        >>> data = await client.get_binary(
        ...     "gopher.floodgap.com",
        ...     "/some/image.gif",
        ... )
        >>> with open("image.gif", "wb") as f:
        ...     f.write(data)
    """
    response = await self.get(
        host=host,
        selector=selector,
        port=port,
        item_type=ItemType.BINARY,
    )

    return response.raw_body or b""

get_attributes async 

get_attributes(
    host: str, selector: str, port: int = DEFAULT_PORT
) -> GopherAttributes

Get Gopher+ attributes for a resource.

Parameters:

Name Type Description Default
host str

The Gopher server hostname.

 required
selector str

The resource selector (path).

 required
port int

The server port. Default is 70.

 DEFAULT_PORT

Returns:

Type Description
GopherAttributes

A GopherAttributes object with metadata.

Raises:

Type Description
TimeoutError

If the request times out.
ConnectionError

If the connection fails.

Examples:

>>> attrs = await client.get_attributes(
...     "gopher.example.com",
...     "/about",
... )
>>> print(attrs.abstract)
Source code in src/mototli/client/session.py 
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
async def get_attributes(
    self,
    host: str,
    selector: str,
    port: int = DEFAULT_PORT,
) -> GopherAttributes:
    """Get Gopher+ attributes for a resource.

    Args:
        host: The Gopher server hostname.
        selector: The resource selector (path).
        port: The server port. Default is 70.

    Returns:
        A GopherAttributes object with metadata.

    Raises:
        asyncio.TimeoutError: If the request times out.
        ConnectionError: If the connection fails.

    Examples:
        >>> attrs = await client.get_attributes(
        ...     "gopher.example.com",
        ...     "/about",
        ... )
        >>> print(attrs.abstract)
    """
    request = GopherRequest(
        selector=selector,
        request_type=RequestType.ATTRIBUTES,
    )

    raw_bytes = await self._send_raw_request(
        host=host,
        port=port,
        request=request,
        is_directory=False,
        use_gopher_plus=True,
    )

    return GopherAttributes.parse(raw_bytes.decode("utf-8", errors="replace"))

get_with_view async 

get_with_view(
    host: str,
    selector: str,
    view_type: str,
    port: int = DEFAULT_PORT,
) -> GopherResponse

Get a specific view of a Gopher+ resource.

Parameters:

Name Type Description Default
host str

The Gopher server hostname.

 required
selector str

The resource selector (path).

 required
view_type str

The MIME type of the view to request.

 required
port int

The server port. Default is 70.

 DEFAULT_PORT

Returns:

Type Description
GopherResponse

A GopherResponse with the requested view.

Raises:

Type Description
TimeoutError

If the request times out.
ConnectionError

If the connection fails.

Examples:

>>> response = await client.get_with_view(
...     "gopher.example.com",
...     "/document",
...     "text/plain",
... )
Source code in src/mototli/client/session.py 
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
async def get_with_view(
    self,
    host: str,
    selector: str,
    view_type: str,
    port: int = DEFAULT_PORT,
) -> GopherResponse:
    """Get a specific view of a Gopher+ resource.

    Args:
        host: The Gopher server hostname.
        selector: The resource selector (path).
        view_type: The MIME type of the view to request.
        port: The server port. Default is 70.

    Returns:
        A GopherResponse with the requested view.

    Raises:
        asyncio.TimeoutError: If the request times out.
        ConnectionError: If the connection fails.

    Examples:
        >>> response = await client.get_with_view(
        ...     "gopher.example.com",
        ...     "/document",
        ...     "text/plain",
        ... )
    """
    request = GopherRequest(
        selector=selector,
        request_type=RequestType.PLUS,
        view_type=view_type,
    )

    return await self._send_request(
        host=host,
        port=port,
        request=request,
        is_directory=False,
    )

Usage Examples

Basic Usage

import asyncio
from mototli.client import GopherClient

async def main():
    async with GopherClient(timeout=30.0) as client:
        response = await client.get("gopher.floodgap.com", "/")
        for item in response.items:
            print(f"[{item.item_type.value}] {item.display_text}")

asyncio.run(main())

Fetching Text

async with GopherClient() as client:
    response = await client.get_text("gopher.floodgap.com", "/gopher/welcome")
    print(response.content.decode())

Fetching Binary

async with GopherClient() as client:
    response = await client.get_binary("gopher.floodgap.com", "/file.zip")
    with open("file.zip", "wb") as f:
        f.write(response.content)

Gopher+ Attributes

async with GopherClient() as client:
    attrs = await client.get_attributes("gopher.floodgap.com", "/gopher")
    print(f"Admin: {attrs.admin.name}")
    print(f"Views: {[v.content_type for v in attrs.views]}")

Search Queries

async with GopherClient() as client:
    response = await client.get(
        "gopher.floodgap.com",
        "/v2/vs",
        search_query="python"
    )
    for item in response.items:
        print(item.display_text)

Error Handling

try:
    async with GopherClient(timeout=10.0) as client:
        response = await client.get("example.com", "/")
except TimeoutError:
    print("Connection timed out")
except ConnectionRefusedError:
    print("Connection refused")
except OSError as e:
    print(f"Network error: {e}")

See Also