tftp.client

tftp.client

The clients: :class:TFTPClient (blocking) and :class:AsyncTFTPClient (asyncio).

Siblings over one private base: the same arguments, the same options and the same results, one waiting on sockets and one on the event loop. The one-shot functions :func:download and :func:upload use the blocking client.

AsyncSink

Bases: Protocol

Where a download writes to: await write(data) takes one chunk.

AsyncSource

Bases: Protocol

Where an upload reads from: await read(n) returns up to n bytes, b"" at the end.

AsyncTFTPClient(host, port=69, *, timeout=1.0, retries=5, blksize=1428, windowsize=None, tsize=True, rollover=None, timeout_option=True, family=0, src=None, fallback=True, dally=False, backoff=2.0, max_timeout=None, deadline=None, max_size=None, strict_source=True, utimeout=False, extra_options=None, registry=None, on_negotiated=None, trace=None)

Bases: _ClientBase

:class:tftp.TFTPClient for asyncio: the same arguments, coroutine methods.

A source or destination is a path (or bytes, to upload) -- local file I/O happens on the loop, which is fine for files and wrong for anything slow -- or an asynchronous stream: an object with async read(n) to upload from, an object with async write(data) to download into. Those are driven with backpressure: a slow sink slows the transfer instead of filling memory. The client never closes a stream it was given.

Source code in src/tftp/client/_core.py

def __init__(
    self,
    host: "HostLike",
    port: int = 69,
    *,
    timeout: float = 1.0,
    retries: int = 5,
    blksize: Union[int, str, None] = 1428,
    windowsize: Optional[int] = None,
    tsize: bool = True,
    rollover: Optional[int] = None,
    timeout_option: bool = True,
    family: int = 0,
    src: Optional[Tuple["HostLike", int]] = None,
    fallback: bool = True,
    dally: bool = False,
    backoff: float = 2.0,
    max_timeout: Optional[float] = None,
    deadline: Optional[float] = None,
    max_size: Optional[int] = None,
    strict_source: bool = True,
    utimeout: bool = False,
    extra_options: Optional[Mapping[str, object]] = None,
    registry: Optional[OptionRegistry] = None,
    on_negotiated: Optional[Callable[[Negotiated, Tuple[Any, ...]], Any]] = None,
    trace: Optional[Callable[[PacketEvent], Any]] = None,
) -> None:
    check_int("port", port, 1, 65535)
    check_int("retries", retries, 0)
    check_seconds("timeout", timeout)
    if max_timeout is not None:
        check_seconds("max_timeout", max_timeout)
        if max_timeout < timeout:
            raise ValueError("max_timeout must be at least timeout (%s), got %s" % (timeout, max_timeout))
    if deadline is not None:
        check_seconds("deadline", deadline, finite=False)
    if max_size is not None:
        check_int("max_size", max_size, 1)
    check_seconds("backoff", backoff, minimum=1.0)
    check_family(family)
    src = check_source(src)
    # Validate the options now rather than on the first transfer.
    request_options(
        blksize=_blksize_number(blksize),
        windowsize=windowsize,
        rollover=rollover,
        extra=extra_options,
    )
    self.host = host
    self.port = port
    self.timeout = timeout
    self.retries = retries
    self.blksize = blksize
    self.windowsize = windowsize
    self.tsize = tsize
    self.rollover = rollover
    self.timeout_option = timeout_option
    self.family = family
    self.src = src
    self.fallback = fallback
    self.dally = dally
    self.backoff = backoff
    self.max_timeout = max_timeout if max_timeout is not None else timeout * 8
    self.deadline = deadline
    self.max_size = max_size
    self.strict_source = strict_source
    self.utimeout = utimeout
    self.extra_options = dict(extra_options or {})
    self.registry = registry
    self.on_negotiated = on_negotiated
    self.trace = trace
    self._trace_guard: Optional[HookGuard] = None

download(filename, dst, *, mode='octet', progress=None, max_size=None) async

Fetch filename into dst: a path, or an object with async write(data).

As :meth:tftp.TFTPClient.download: a path is replaced only when the transfer succeeds, and max_size overrides the client's.

Source code in src/tftp/client/_asyncio.py

async def download(
    self,
    filename: str,
    dst: Union[str, "os.PathLike[str]", AsyncSink],
    *,
    mode: str = "octet",
    progress: Optional[ProgressFunction] = None,
    max_size: Optional[int] = None,
) -> TransferResult:
    """Fetch ``filename`` into ``dst``: a path, or an object with ``async write(data)``.

    As :meth:`tftp.TFTPClient.download`: a path is replaced only when the
    transfer succeeds, and ``max_size`` overrides the client's.
    """
    mode = _mode(mode)
    limit = self._limit(max_size)
    if isinstance(dst, (str, os.PathLike)):
        target = _PathSink(dst)
        try:
            result = await self._download(filename, target.sink, mode, progress, limit, bridged=False)
            target.commit()
        except BaseException:
            target.abort()
            raise
        return result
    return await self._download(filename, dst, mode, progress, limit, bridged=True)

listdir(dirname='', *, max_size=LISTING_LIMIT) async

:meth:tftp.TFTPClient.listdir, as a coroutine.

Source code in src/tftp/client/_asyncio.py

async def listdir(self, dirname: str = "", *, max_size: Optional[int] = LISTING_LIMIT) -> List[ListEntry]:
    """:meth:`tftp.TFTPClient.listdir`, as a coroutine."""
    lister = self._lister()
    sink = io.BytesIO()
    try:
        await lister._download(
            dirname or ".", sink, "octet", None, lister._limit(max_size), bridged=False
        )
    except _NotListing:
        raise NotADirectoryError(errno.ENOTDIR, "not a directory", dirname) from None
    return listing.loads(sink.getvalue())

size(filename, *, mode='octet') async

:meth:tftp.TFTPClient.size, without blocking the loop (it runs in the executor).

Source code in src/tftp/client/_asyncio.py

async def size(self, filename: str, *, mode: str = "octet") -> Optional[int]:
    """:meth:`tftp.TFTPClient.size`, without blocking the loop (it runs in the executor)."""
    loop = asyncio.get_running_loop()
    expires = self._expires(time.monotonic())
    return await loop.run_in_executor(None, lambda: self._size(filename, mode, expires))

stat(filename, *, mode='octet') async

:meth:tftp.TFTPClient.stat, without blocking the loop (it runs in the executor).

Source code in src/tftp/client/_asyncio.py

async def stat(self, filename: str, *, mode: str = "octet") -> RemoteStat:
    """:meth:`tftp.TFTPClient.stat`, without blocking the loop (it runs in the executor)."""
    loop = asyncio.get_running_loop()
    expires = self._expires(time.monotonic())
    return await loop.run_in_executor(None, lambda: self._stat(filename, mode, expires))

stream(filename, *, mode='octet', buffer=1 << 20) async

Yield filename's contents as it arrives, with backpressure.

At most buffer bytes are held: half in the transfer's own buffer and at most as much again waiting for the consumer. When the consumer is slow, acknowledgements are held back and the server waits. Closing the generator early ends the transfer (the server is sent ERROR 0).

Source code in src/tftp/client/_asyncio.py

async def stream(
    self, filename: str, *, mode: str = "octet", buffer: int = 1 << 20
) -> AsyncIterator[bytes]:
    """Yield ``filename``'s contents as it arrives, with backpressure.

    At most ``buffer`` bytes are held: half in the transfer's own buffer
    and at most as much again waiting for the consumer. When the consumer
    is slow, acknowledgements are held back and the server waits. Closing
    the generator early ends the transfer (the server is sent ERROR 0).
    """
    queue: "asyncio.Queue[Optional[bytes]]" = asyncio.Queue()
    taken = asyncio.Event()

    class _Sink:
        async def write(self, data: bytes) -> None:
            taken.clear()
            queue.put_nowait(data)
            await taken.wait()

    task = asyncio.ensure_future(
        self._download(
            filename,
            _Sink(),
            _mode(mode),
            None,
            self._limit(None),
            bridged=True,
            capacity=max(buffer // 2, 1),
        )
    )
    task.add_done_callback(lambda _: queue.put_nowait(None))
    try:
        while True:
            chunk = await queue.get()
            if chunk is None:
                break
            taken.set()
            yield chunk
        await task  # re-raise a transfer error
    finally:
        if not task.done():
            task.cancel()
            await asyncio.wait({task})

upload(filename, src, *, mode='octet', progress=None) async

Send src: a path, bytes, or an object with async read(n).

Source code in src/tftp/client/_asyncio.py

async def upload(
    self,
    filename: str,
    src: Union[str, "os.PathLike[str]", bytes, bytearray, memoryview, AsyncSource],
    *,
    mode: str = "octet",
    progress: Optional[ProgressFunction] = None,
) -> TransferResult:
    """Send ``src``: a path, bytes, or an object with ``async read(n)``."""
    mode = _mode(mode)
    if isinstance(src, (str, os.PathLike)):
        with open(os.fspath(src), "rb") as fileobj:
            return await self._upload(filename, fileobj, mode, progress, bridged=False)
    if isinstance(src, (bytes, bytearray, memoryview)):
        return await self._upload(filename, io.BytesIO(bytes(src)), mode, progress, bridged=False)
    return await self._upload(filename, src, mode, progress, bridged=True)

RemoteStat

Bases: NamedTuple

What :meth:TFTPClient.stat learnt about a name on the server.

size is None when the server reports none (and for a directory); mtime (seconds since the epoch) needs a server speaking x-mtime; is_dir needs one speaking x-list -- other servers report a directory as not found.

TFTPClient(host, port=69, *, timeout=1.0, retries=5, blksize=1428, windowsize=None, tsize=True, rollover=None, timeout_option=True, family=0, src=None, fallback=True, dally=False, backoff=2.0, max_timeout=None, deadline=None, max_size=None, strict_source=True, utimeout=False, extra_options=None, registry=None, on_negotiated=None, trace=None)

Bases: _ClientBase

A TFTP client bound to one server.

:param host: server name or address: a string ("name", "10.0.0.1", "[v6]:port"), an ipaddress address or interface, or a netimps.Host. Only a string can carry a port. :param port: server port. :param timeout: seconds before a retransmission. Also requested from the server, as timeout when whole and as utimeout when fractional, unless timeout_option is false. :param retries: retransmissions of one packet before giving up. :param blksize: requested DATA size, 8..65464. None asks for nothing, leaving RFC 1350's 512. The default 1428 fits an Ethernet frame on IPv4 and IPv6 without fragmenting. "mtu" sizes it to the MTU of the interface the route to the server uses (falling back to 1428). :param windowsize: requested RFC 7440 window. None asks for nothing (one packet per ACK). :param tsize: request the transfer size (RRQ) or announce it (WRQ). :param rollover: request block-number rollover to 0 or 1. None asks for nothing; wrapping to 0 is assumed and a server wrapping to 1 is followed. :param family: socket.AF_INET / AF_INET6 to force one, 0 for whatever host resolves to first. :param src: (address, port) to send from; the address as for host. :param fallback: when the server answers a request that carries options with ERROR 8, 4 or 0 (RFC 2347: it "may" be repeated without them), retry once without any. :param dally: after acknowledging the last DATA of a download, keep answering a retransmitted last DATA for one timeout (RFC 1350 section 6). Costs that much time on every download. :param backoff: each consecutive retransmission waits this many times longer (RFC 1123 4.2.3.2); progress resets it to timeout. :param max_timeout: ceiling for the backed-off wait, at least timeout; None is eight times timeout. :param deadline: seconds a whole transfer may take, or None. A relative duration, unlike the engine's deadline attribute. :param max_size: octets a download may bring, or None for no bound. A server announcing more, or sending more than it announced or than this, is sent ERROR 3 and ends the download with :class:TransferTooLargeError (or :class:TFTPProtocolError for more than it announced). :param utimeout: send a fractional timeout as tftp-hpa's utimeout (otherwise a fractional timeout is not requested at all). :param extra_options: further options to request, verbatim (extensions such as blksize2/cookie, or custom ones). Built-in ones are validated in the OACK; custom ones land in negotiated.extra. :param registry: the :class:OptionRegistry used to validate the OACK. :param trace: trace(PacketEvent) for every datagram this client sends or receives (role="client", one session id per transfer). :param on_negotiated: called as on_negotiated(negotiated, peer) once the server has answered the request (OACK, first DATA or ACK 0), before any data moves -- the peer is the server's transfer address. :param strict_source: the first answer must come from the address the request was sent to. False accepts any address, for multi-homed servers that answer from another one (the transfer then locks on to whichever answered first).

A client is not thread-safe; use one per thread.

Source code in src/tftp/client/_core.py

def __init__(
    self,
    host: "HostLike",
    port: int = 69,
    *,
    timeout: float = 1.0,
    retries: int = 5,
    blksize: Union[int, str, None] = 1428,
    windowsize: Optional[int] = None,
    tsize: bool = True,
    rollover: Optional[int] = None,
    timeout_option: bool = True,
    family: int = 0,
    src: Optional[Tuple["HostLike", int]] = None,
    fallback: bool = True,
    dally: bool = False,
    backoff: float = 2.0,
    max_timeout: Optional[float] = None,
    deadline: Optional[float] = None,
    max_size: Optional[int] = None,
    strict_source: bool = True,
    utimeout: bool = False,
    extra_options: Optional[Mapping[str, object]] = None,
    registry: Optional[OptionRegistry] = None,
    on_negotiated: Optional[Callable[[Negotiated, Tuple[Any, ...]], Any]] = None,
    trace: Optional[Callable[[PacketEvent], Any]] = None,
) -> None:
    check_int("port", port, 1, 65535)
    check_int("retries", retries, 0)
    check_seconds("timeout", timeout)
    if max_timeout is not None:
        check_seconds("max_timeout", max_timeout)
        if max_timeout < timeout:
            raise ValueError("max_timeout must be at least timeout (%s), got %s" % (timeout, max_timeout))
    if deadline is not None:
        check_seconds("deadline", deadline, finite=False)
    if max_size is not None:
        check_int("max_size", max_size, 1)
    check_seconds("backoff", backoff, minimum=1.0)
    check_family(family)
    src = check_source(src)
    # Validate the options now rather than on the first transfer.
    request_options(
        blksize=_blksize_number(blksize),
        windowsize=windowsize,
        rollover=rollover,
        extra=extra_options,
    )
    self.host = host
    self.port = port
    self.timeout = timeout
    self.retries = retries
    self.blksize = blksize
    self.windowsize = windowsize
    self.tsize = tsize
    self.rollover = rollover
    self.timeout_option = timeout_option
    self.family = family
    self.src = src
    self.fallback = fallback
    self.dally = dally
    self.backoff = backoff
    self.max_timeout = max_timeout if max_timeout is not None else timeout * 8
    self.deadline = deadline
    self.max_size = max_size
    self.strict_source = strict_source
    self.utimeout = utimeout
    self.extra_options = dict(extra_options or {})
    self.registry = registry
    self.on_negotiated = on_negotiated
    self.trace = trace
    self._trace_guard: Optional[HookGuard] = None

download(filename, dst, *, mode='octet', progress=None, max_size=None)

Fetch filename into dst (a path or a writable binary file).

A path is written to a temporary file beside it, which replaces it when the transfer succeeds: a download that fails leaves what was there. A path that exists and is not a regular file (a device, a pipe) is written in place. max_size overrides the client's. Raises :class:RemoteError, :class:TransferTimeoutError, :class:TransferTooLargeError or :class:TFTPProtocolError; OSError for local failures.

Source code in src/tftp/client/_sync.py

def download(
    self,
    filename: str,
    dst: SinkLike,
    *,
    mode: str = "octet",
    progress: Optional[ProgressFunction] = None,
    max_size: Optional[int] = None,
) -> TransferResult:
    """Fetch ``filename`` into ``dst`` (a path or a writable binary file).

    A path is written to a temporary file beside it, which replaces it when
    the transfer succeeds: a download that fails leaves what was there. A
    path that exists and is not a regular file (a device, a pipe) is
    written in place. ``max_size`` overrides the client's.
    Raises :class:`RemoteError`, :class:`TransferTimeoutError`,
    :class:`TransferTooLargeError` or :class:`TFTPProtocolError`;
    ``OSError`` for local failures.
    """
    mode = _mode(mode)
    limit = self._limit(max_size)
    if isinstance(dst, (str, os.PathLike)):
        target = _PathSink(dst)
        try:
            result = self._download(filename, target.sink, mode, progress, limit)
            target.commit()
        except BaseException:
            target.abort()
            raise
        return result
    return self._download(filename, dst, mode, progress, limit)

get(filename, *, mode='octet', max_size=None)

Fetch filename and return its contents.

Source code in src/tftp/client/_sync.py

def get(self, filename: str, *, mode: str = "octet", max_size: Optional[int] = None) -> bytes:
    """Fetch ``filename`` and return its contents."""
    buffer = io.BytesIO()
    self.download(filename, buffer, mode=mode, max_size=max_size)
    return buffer.getvalue()

listdir(dirname='', *, max_size=LISTING_LIMIT)

The entries of directory dirname ("" is the server's root).

A listing longer than max_size octets (16 MiB) raises :class:TransferTooLargeError.

Needs a server speaking pytftp's x-list extension (tftp.TFTPServer allowing :data:LISTING_OPTIONS); there is no standard way to list in TFTP. Raises NotADirectoryError when the name is a file (the transfer is abandoned at once), and :class:FileNotFound when it does not exist -- which is also what a server without listing support answers for a directory.

Source code in src/tftp/client/_sync.py

def listdir(self, dirname: str = "", *, max_size: Optional[int] = LISTING_LIMIT) -> List[ListEntry]:
    """The entries of directory ``dirname`` (``""`` is the server's root).

    A listing longer than ``max_size`` octets (16 MiB) raises
    :class:`TransferTooLargeError`.

    Needs a server speaking pytftp's ``x-list`` extension (``tftp.TFTPServer``
    allowing :data:`LISTING_OPTIONS`); there is no standard way to list
    in TFTP. Raises ``NotADirectoryError`` when the name is a file (the
    transfer is abandoned at once), and :class:`FileNotFound` when it does
    not exist -- which is also what a server without listing support
    answers for a directory.
    """
    lister = self._lister()
    sink = io.BytesIO()
    try:
        lister.download(dirname or ".", sink, max_size=max_size)
    except _NotListing:
        raise NotADirectoryError(errno.ENOTDIR, "not a directory", dirname) from None
    return listing.loads(sink.getvalue())

path(*segments, mode='octet')

A :class:tftp.path.TFTPPath on this server (needs the path extra).

Source code in src/tftp/client/_sync.py

def path(self, *segments: Any, mode: str = "octet") -> TFTPPath:
    """A :class:`tftp.path.TFTPPath` on this server (needs the ``path`` extra)."""
    from ..path._local import TFTPPath

    return TFTPPath(*segments, client=self, mode=mode)

put(filename, data, *, mode='octet')

Upload data as filename.

Source code in src/tftp/client/_sync.py

def put(self, filename: str, data: bytes, *, mode: str = "octet") -> TransferResult:
    """Upload ``data`` as ``filename``."""
    return self.upload(filename, data, mode=mode)

size(filename, *, mode='octet')

The size of filename on the server, without transferring it.

Sends an RRQ asking only for tsize and abandons the transfer as soon as the server answers (ERROR 8 to an OACK, as RFC 2347 lets a client refuse one). Returns None when the server does not report sizes -- unless the whole file fits in the first block, whose length then is the size. Raises :class:RemoteError (FileNotFound...) like a download would.

Source code in src/tftp/client/_sync.py

def size(self, filename: str, *, mode: str = "octet") -> Optional[int]:
    """The size of ``filename`` on the server, without transferring it.

    Sends an RRQ asking only for ``tsize`` and abandons the transfer as
    soon as the server answers (ERROR 8 to an OACK, as RFC 2347 lets a
    client refuse one). Returns ``None`` when the server does not report
    sizes -- unless the whole file fits in the first block, whose length
    then is the size. Raises :class:`RemoteError` (``FileNotFound``...)
    like a download would.
    """
    return self._size(filename, mode, self._expires(time.monotonic()))

stat(filename, *, mode='octet')

Size, modification time and kind of filename, in one probe.

Like :meth:size, with x-mtime and x-list asked for as well: a pytftp server allowing them reports the time and recognises a directory; any other server ignores them (RFC 2347). A server that refuses the request over its options (ERROR 8) is probed again for the size alone, when fallback is on.

Source code in src/tftp/client/_sync.py

def stat(self, filename: str, *, mode: str = "octet") -> RemoteStat:
    """Size, modification time and kind of ``filename``, in one probe.

    Like :meth:`size`, with ``x-mtime`` and ``x-list`` asked for as well:
    a pytftp server allowing them reports the time and recognises a
    directory; any other server ignores them (RFC 2347). A server that
    refuses the request over its options (ERROR 8) is probed again for
    the size alone, when ``fallback`` is on.
    """
    return self._stat(filename, mode, self._expires(time.monotonic()))

upload(filename, src, *, mode='octet', progress=None)

Send src (a path, a readable binary file or bytes) as filename.

Source code in src/tftp/client/_sync.py

def upload(
    self,
    filename: str,
    src: SourceLike,
    *,
    mode: str = "octet",
    progress: Optional[ProgressFunction] = None,
) -> TransferResult:
    """Send ``src`` (a path, a readable binary file or bytes) as ``filename``."""
    mode = _mode(mode)
    if isinstance(src, (str, os.PathLike)):
        with open(os.fspath(src), "rb") as fileobj:
            return self._upload(filename, fileobj, mode, progress)
    if isinstance(src, (bytes, bytearray, memoryview)):
        return self._upload(filename, io.BytesIO(bytes(src)), mode, progress)
    return self._upload(filename, src, mode, progress)

__getattr__(name)

Bind :class:AsyncTFTPClient on first use, so importing the package does not import asyncio.

Source code in src/tftp/client/__init__.py

def __getattr__(name: str) -> Any:
    """Bind :class:`AsyncTFTPClient` on first use, so importing the package does not import asyncio."""
    if name == "AsyncTFTPClient":
        from ._asyncio import AsyncTFTPClient

        globals()[name] = AsyncTFTPClient
        return AsyncTFTPClient
    raise AttributeError("module %r has no attribute %r" % (__name__, name))

download(host, filename, dst, /, *, port=69, mode='octet', progress=None, **client_options)

One-shot :meth:TFTPClient.download; client_options go to :class:TFTPClient.

Source code in src/tftp/client/_sync.py

def download(
    host: "HostLike",
    filename: str,
    dst: SinkLike,
    /,
    *,
    port: int = 69,
    mode: str = "octet",
    progress: Optional[ProgressFunction] = None,
    **client_options: Any,
) -> TransferResult:
    """One-shot :meth:`TFTPClient.download`; ``client_options`` go to :class:`TFTPClient`."""
    return TFTPClient(host, port, **client_options).download(filename, dst, mode=mode, progress=progress)

upload(host, filename, src, /, *, port=69, mode='octet', progress=None, **client_options)

One-shot :meth:TFTPClient.upload; client_options go to :class:TFTPClient.

Source code in src/tftp/client/_sync.py

def upload(
    host: "HostLike",
    filename: str,
    src: SourceLike,
    /,
    *,
    port: int = 69,
    mode: str = "octet",
    progress: Optional[ProgressFunction] = None,
    **client_options: Any,
) -> TransferResult:
    """One-shot :meth:`TFTPClient.upload`; ``client_options`` go to :class:`TFTPClient`."""
    return TFTPClient(host, port, **client_options).upload(filename, src, mode=mode, progress=progress)