strukto-ai--mirage/python/mirage/commands/builtin/generic_bind/adapter.py

# ========= Copyright 2026 @ Strukto.AI All Rights Reserved. =========
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
#     http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
# ========= Copyright 2026 @ Strukto.AI All Rights Reserved. =========

import functools
from collections.abc import Callable
from dataclasses import dataclass
from typing import NamedTuple

from mirage.accessor.base import Accessor
from mirage.cache.index import IndexCacheStore
from mirage.types import PathSpec
from mirage.utils.glob_walk import resolve_glob_with


def with_index(fn: Callable | None,
               index: IndexCacheStore | None) -> Callable | None:
    """Bind the runtime cache index into a read op for the generics.

    A generic command calls its injected reader as ``read(accessor, path)``
    with no index, but index-backed backends (gdrive, gmail, slack, ...)
    resolve a path to its real id through the index, so the bound index must
    travel with the reader. Harmless for backends that ignore it. ``None``
    passes through so a backend/test can still opt out of streaming.

    Args:
        fn (Callable | None): the read op, or None to opt out of streaming.
        index (IndexCacheStore | None): the per-call cache index.
    """
    if fn is None:
        return None
    return functools.partial(fn, index=index)


class Builder(NamedTuple):
    name: str
    fn: Callable
    provision: Callable | None = None
    write: bool = False
    aggregate: Callable | None = None
    read: bool = False


def make_resolve_glob(readdir: Callable,
                      max_glob_matches: int | None = None) -> Callable:
    """Build a resolve_glob generic over a backend's readdir.

    Args:
        readdir (Callable): backend readdir ``(accessor, path, index)``.
        max_glob_matches (int | None): cap on matches per pattern before
            truncation.
    """

    async def resolve_glob(accessor: Accessor, paths: list[PathSpec],
                           index: IndexCacheStore | None) -> list[PathSpec]:
        return await resolve_glob_with(readdir, accessor, paths, index,
                                       max_glob_matches)

    return resolve_glob


@dataclass(frozen=True)
class CommandIO:
    readdir: Callable
    read_bytes: Callable
    read_stream: Callable
    stat: Callable
    is_mounted: Callable
    local: bool = True
    max_glob_matches: int | None = None
    write: Callable | None = None
    exists: Callable | None = None
    mkdir: Callable | None = None
    unlink: Callable | None = None
    rmdir: Callable | None = None
    rm_r: Callable | None = None
    rename: Callable | None = None
    copy: Callable | None = None
    dir_copy: Callable | None = None
    create: Callable | None = None
    truncate: Callable | None = None
    find: Callable | None = None
    is_dir_name: Callable | None = None
    du_total: Callable | None = None
    du_all: Callable | None = None

    @property
    def resolve_glob(self) -> Callable:
        return make_resolve_glob(self.readdir, self.max_glob_matches)