Source code for bacommon.bacloud

# Released under the MIT License. See LICENSE for details.
#
"""Functionality related to the bacloud tool."""

from __future__ import annotations

from dataclasses import dataclass
from typing import TYPE_CHECKING, Annotated

from efro.dataclassio import ioprepped, IOAttrs

if TYPE_CHECKING:
    pass

# Version is sent to the master-server with all commands. Can be incremented
# if we need to change behavior server-side to go along with client changes.
BACLOUD_VERSION = 13




def asset_file_cache_path(filehash: str) -> str:
    """Given a sha256 hex file hash, return a storage path."""

    # We expect a 64 byte hex str with only lowercase letters and
    # numbers. Note to self: I considered base64 hashes to save space
    # but then remembered that lots of filesystems out there ignore case
    # so that would not end well.
    assert len(filehash) == 64
    assert filehash.islower()
    assert filehash.isalnum()

    # Split into a few levels of directories to keep directory listings
    # and operations reasonable. This will give 256 top level dirs, each
    # with 256 subdirs. So if we have 65,536 files in our cache then
    # dirs will average 1 file each. That seems like a reasonable spread
    # I think.
    return f'{filehash[:2]}/{filehash[2:4]}/{filehash[4:]}'





@ioprepped
@dataclass
class RequestData:
    """Request sent to bacloud server."""

    command: str
    token: str | None
    payload: dict
    tzoffset: float
    isatty: bool





@ioprepped
@dataclass
class ResponseData:
    """Response sent from the bacloud server to the client."""



    @ioprepped
    @dataclass
    class Downloads:
        """Info about downloads included in a response."""



        @ioprepped
        @dataclass
        class Entry:
            """Individual download."""

            path: str

            #: Args include with this particular request (combined with
            #: baseargs).
            args: dict[str, str]


            # TODO: could add a hash here if we want the client to
            # verify hashes.

        #: If present, will be prepended to all entry paths via os.path.join.
        basepath: str | None

        #: Server command that should be called for each download. The
        #: server command is expected to respond with a downloads_inline
        #: containing a single 'default' entry. In the future this may
        #: be expanded to a more streaming-friendly process.
        cmd: str

        #: Args that should be included with all download requests.
        baseargs: dict[str, str]

        #: Everything that should be downloaded.
        entries: list[Entry]


    #: If present, client should print this message before any other
    #: response processing (including error handling) occurs.
    message: str | None = None

    #: End arg for message print() call.
    message_end: str = '\n'

    #: If present, client should abort with this error message.
    error: str | None = None

    #: How long to wait before proceeding with remaining response (can
    #: be useful when waiting for server progress in a loop).
    delay_seconds: float = 0.0

    #: If present, a token that should be stored client-side and passed
    #: with subsequent commands.
    login: str | None = None

    #: If True, any existing client-side token should be discarded.
    logout: bool = False

    #: If present, client should generate a manifest of this dir.
    #: It should be added to end_command args as 'manifest'.
    dir_manifest: str | None = (
        None
    )

    #: If present, client should upload the requested files (arg1)
    #: individually to a server command (arg2) with provided args (arg3).
    uploads: tuple[list[str], str, dict] | None = None

    #: If present, a list of pathnames that should be gzipped
    #: and uploaded to an 'uploads_inline' bytes dict in end_command args.
    #: This should be limited to relatively small files.
    uploads_inline: list[str] | None = None

    #: If present, file paths that should be deleted on the client.
    deletes: list[str] | None = None

    #: If present, describes files the client should individually
    #: request from the server if not already present on the client.
    downloads: Downloads | None = None

    #: If present, pathnames mapped to gzipped data to
    #: be written to the client. This should only be used for relatively
    #: small files as they are all included inline as part of the response.
    downloads_inline: dict[str, bytes] | None = None

    #: If present, all empty dirs under this one should be removed.
    dir_prune_empty: str | None = None

    #: If present, url to display to the user.
    open_url: str | None = None

    #: If present, a line of input is read and placed into
    #: end_command args as 'input'. The first value is the prompt printed
    #: before reading and the second is whether it should be read as a
    #: password (without echoing to the terminal).
    input_prompt: tuple[str, bool] | None = None

    #: If present, a message that should be printed after all other
    #: response processing is done.
    end_message: str | None = (
        None
    )

    #: End arg for end_message print() call.
    end_message_end: str = '\n'

    #: If present, this command is run with these args at the end
    #: of response processing.
    end_command: tuple[str, dict] | None = None



# Docs-generation hack; import some stuff that we likely only forward-declared
# in our actual source code so that docs tools can find it.
from typing import (Coroutine, Any, Literal, Callable,
  Generator, Awaitable, Sequence, Self)
import asyncio
from concurrent.futures import Future