fsspeckit.core API Reference

core

Core filesystem functionality for fsspeckit.

Classes

fsspeckit.core.GitLabFileSystem

GitLabFileSystem(
    base_url: str = "https://gitlab.com",
    project_id: str | int | None = None,
    project_name: str | None = None,
    ref: str = "main",
    token: str | None = None,
    api_version: str = "v4",
    timeout: float = 30.0,
    max_pages: int = 1000,
    **kwargs: Any,
)

Bases: AbstractFileSystem

Filesystem interface for GitLab repositories.

Provides read-only access to files in GitLab repositories, including: - Public and private repositories - Self-hosted GitLab instances - Branch/tag/commit selection - Token-based authentication

Attributes:

Name	Type	Description
protocol	str	Always "gitlab"
base_url	str	GitLab instance URL
project_id	str	Project ID
project_name	str	Project name/path
ref	str	Git reference (branch, tag, commit)
token	str	Access token
api_version	str	API version

Example

# Public repository
fs = GitLabFileSystem(
    project_name="group/project",
    ref="main",
)
files = fs.ls("/")

# Private repository with token
fs = GitLabFileSystem(
    project_id="12345",
    token="glpat_xxxx",
    ref="develop",
)
content = fs.cat("README.md")

Initialize GitLab filesystem.

Parameters:

Name	Type	Description	Default
base_url	str	GitLab instance URL	'https://gitlab.com'
project_id	str | int | None	Project ID number	None
project_name	str | None	Project name/path (alternative to project_id)	None
ref	str	Git reference (branch, tag, or commit SHA)	'main'
token	str | None	GitLab personal access token	None
api_version	str	API version to use	'v4'
timeout	float	Request timeout in seconds (must be positive, max 3600)	30.0
max_pages	int	Maximum number of pages to fetch (default 1000, max 10000)	1000
**kwargs	Any	Additional arguments	{}

Source code in src/fsspeckit/core/filesystem/gitlab.py

def __init__(
    self,
    base_url: str = "https://gitlab.com",
    project_id: str | int | None = None,
    project_name: str | None = None,
    ref: str = "main",
    token: str | None = None,
    api_version: str = "v4",
    timeout: float = 30.0,
    max_pages: int = 1000,
    **kwargs: Any,
):
    """Initialize GitLab filesystem.

    Args:
        base_url: GitLab instance URL
        project_id: Project ID number
        project_name: Project name/path (alternative to project_id)
        ref: Git reference (branch, tag, or commit SHA)
        token: GitLab personal access token
        api_version: API version to use
        timeout: Request timeout in seconds (must be positive, max 3600)
        max_pages: Maximum number of pages to fetch (default 1000, max 10000)
        **kwargs: Additional arguments
    """
    super().__init__(**kwargs)

    self.base_url = base_url.rstrip("/")
    self.project_id = project_id
    self.project_name = project_name
    self.ref = ref
    self.token = token
    self.api_version = api_version

    # Input validation
    if timeout <= 0:
        raise ValueError("timeout must be a positive number")
    if timeout > 3600:
        raise ValueError("timeout must not exceed 3600 seconds")
    self.timeout = timeout

    if max_pages <= 0:
        raise ValueError("max_pages must be a positive integer")
    if max_pages > 10000:
        raise ValueError("max_pages must not exceed 10000")
    self.max_pages = max_pages

    if not project_id and not project_name:
        raise ValueError("Either project_id or project_name must be provided")

    # Create a shared requests session with timeout
    self._session = requests.Session()
    if self.token:
        self._session.headers["PRIVATE-TOKEN"] = self.token

    # Track closed state for resource cleanup
    self._closed = False

Functions

fsspeckit.core.GitLabFileSystem.__del__

__del__() -> None

Destructor to ensure cleanup when object is garbage collected.

Source code in src/fsspeckit/core/filesystem/gitlab.py

def __del__(self) -> None:
    """Destructor to ensure cleanup when object is garbage collected."""
    try:
        self.close()
    except Exception:
        # Silently ignore errors in destructor
        pass

fsspeckit.core.GitLabFileSystem.cat_file

cat_file(path: str, **kwargs: Any) -> bytes

Get file content.

Parameters:

Name	Type	Description	Default
path	str	File path	required
**kwargs	Any	Additional arguments	{}

Returns:

Type	Description
bytes	File content

Raises:

Type	Description
HTTPError	If file not found or other HTTP error

Source code in src/fsspeckit/core/filesystem/gitlab.py

def cat_file(self, path: str, **kwargs: Any) -> bytes:
    """Get file content.

    Args:
        path: File path
        **kwargs: Additional arguments

    Returns:
        File content

    Raises:
        requests.HTTPError: If file not found or other HTTP error
    """
    params = {"ref": self.ref}

    # URL-encode the file path
    encoded_path = urllib.parse.quote(path.lstrip("/"), safe="")

    response = self._make_request(f"repository/files/{encoded_path}", params)
    data = response.json()

    import base64

    return base64.b64decode(data["content"])

fsspeckit.core.GitLabFileSystem.close

close() -> None

Close the filesystem and cleanup resources.

This method closes the internal requests session to prevent resource leaks. It is safe to call this method multiple times.

Source code in src/fsspeckit/core/filesystem/gitlab.py

def close(self) -> None:
    """Close the filesystem and cleanup resources.

    This method closes the internal requests session to prevent resource leaks.
    It is safe to call this method multiple times.
    """
    if not self._closed:
        logger.debug("Closing GitLabFileSystem and cleaning up resources")
        self._session.close()
        self._closed = True

fsspeckit.core.GitLabFileSystem.exists

exists(path: str, **kwargs: Any) -> bool

Check if file exists.

Parameters:

Name	Type	Description	Default
path	str	File path	required
**kwargs	Any	Additional arguments	{}

Returns:

Type	Description
bool	True if file exists

Source code in src/fsspeckit/core/filesystem/gitlab.py

def exists(self, path: str, **kwargs: Any) -> bool:
    """Check if file exists.

    Args:
        path: File path
        **kwargs: Additional arguments

    Returns:
        True if file exists
    """
    try:
        self.info(path, **kwargs)
        return True
    except requests.HTTPError as e:
        if e.response and e.response.status_code == 404:
            return False
        # Re-raise for other HTTP errors
        raise
    except requests.RequestException:
        # Re-raise for other request errors
        raise

fsspeckit.core.GitLabFileSystem.info

info(path: str, **kwargs: Any) -> dict

Get file information.

Parameters:

Name	Type	Description	Default
path	str	File path	required
**kwargs	Any	Additional arguments	{}

Returns:

Type	Description
dict	File information

Raises:

Type	Description
HTTPError	If file not found or other HTTP error

Source code in src/fsspeckit/core/filesystem/gitlab.py

def info(self, path: str, **kwargs: Any) -> dict:
    """Get file information.

    Args:
        path: File path
        **kwargs: Additional arguments

    Returns:
        File information

    Raises:
        requests.HTTPError: If file not found or other HTTP error
    """
    params = {"ref": self.ref}

    # URL-encode the file path
    encoded_path = urllib.parse.quote(path.lstrip("/"), safe="")

    response = self._make_request(f"repository/files/{encoded_path}", params)
    return response.json()

fsspeckit.core.GitLabFileSystem.ls

ls(
    path: str = "", detail: bool = False, **kwargs: Any
) -> list[Any] | Any

List files in repository with pagination support.

Parameters:

Name	Type	Description	Default
path	str	Directory path	''
detail	bool	Whether to return detailed information	False
**kwargs	Any	Additional arguments	{}

Returns:

Type	Description
list[Any] | Any	List of files

Source code in src/fsspeckit/core/filesystem/gitlab.py

def ls(
    self, path: str = "", detail: bool = False, **kwargs: Any
) -> list[Any] | Any:
    """List files in repository with pagination support.

    Args:
        path: Directory path
        detail: Whether to return detailed information
        **kwargs: Additional arguments

    Returns:
        List of files
    """
    all_files = []
    page = 1
    per_page = 100
    pages_fetched = 0

    while pages_fetched < self.max_pages:
        params = {"ref": self.ref, "per_page": per_page, "page": page}

        if path:
            params["path"] = path.lstrip("/")

        try:
            response = self._make_request("repository/tree", params)
            files = response.json()

            if not files:
                # No more pages
                break

            all_files.extend(files)
            pages_fetched += 1

            # Check for pagination headers
            next_page = response.headers.get("X-Next-Page")
            if not next_page:
                # No more pages
                break

            # Try to parse the next page number
            try:
                page = int(next_page)
            except (ValueError, TypeError):
                # Malformed pagination header
                logger.warning(
                    "Malformed X-Next-Page header: '%s', stopping pagination at page %d",
                    next_page,
                    page,
                )
                break

        except requests.RequestException:
            # If we have some files already, return what we have
            if all_files:
                logger.warning(
                    "GitLab API request failed for page %d, returning %d files from previous pages",
                    page,
                    len(all_files),
                )
                break
            else:
                # Re-raise if no files collected yet
                raise
    else:
        # Loop ended due to max_pages limit
        logger.warning(
            "Reached maximum pages limit (%d), returning %d files",
            self.max_pages,
            len(all_files),
        )

    if detail:
        return all_files
    else:
        return [item["name"] for item in all_files]

fsspeckit.core.MonitoredSimpleCacheFileSystem

MonitoredSimpleCacheFileSystem(
    fs: AbstractFileSystem | None = None,
    cache_storage: str = "~/.cache/fsspec",
    verbose: bool = False,
    **kwargs: Any,
)

Bases: SimpleCacheFileSystem

Simple cache filesystem with monitoring and logging.

This filesystem wraps another filesystem and caches files locally. It provides monitoring capabilities and verbose logging of cache operations.

Parameters:

Name	Type	Description	Default
fs	AbstractFileSystem | None	Underlying filesystem to cache. If None, creates a local filesystem.	None
cache_storage	str	Cache storage location(s). Can be string path or list of paths.	'~/.cache/fsspec'
verbose	bool	Whether to enable verbose logging of cache operations.	False
**kwargs	Any	Additional arguments passed to SimpleCacheFileSystem.	{}

Example

# Cache S3 filesystem
s3_fs = filesystem("s3")
cached = MonitoredSimpleCacheFileSystem(
    fs=s3_fs,
    cache_storage="/tmp/s3_cache",
    verbose=True,
)

Initialize monitored cache filesystem.

Parameters:

Name	Type	Description	Default
fs	AbstractFileSystem | None	Underlying filesystem to cache. If None, creates a local filesystem.	None
cache_storage	str	Cache storage location(s). Can be string path or list of paths.	'~/.cache/fsspec'
verbose	bool	Whether to enable verbose logging of cache operations.	False
**kwargs	Any	Additional arguments passed to SimpleCacheFileSystem.	{}

Source code in src/fsspeckit/core/filesystem/cache.py

def __init__(
    self,
    fs: fsspec.AbstractFileSystem | None = None,
    cache_storage: str = "~/.cache/fsspec",
    verbose: bool = False,
    **kwargs: Any,
):
    """Initialize monitored cache filesystem.

    Args:
        fs: Underlying filesystem to cache. If None, creates a local filesystem.
        cache_storage: Cache storage location(s). Can be string path or list of paths.
        verbose: Whether to enable verbose logging of cache operations.
        **kwargs: Additional arguments passed to SimpleCacheFileSystem.
    """
    self._verbose = verbose

    super().__init__(fs=fs, cache_storage=cache_storage, **kwargs)
    self._mapper = FileNameCacheMapper(cache_storage)

    if self._verbose:
        logger.info(f"Initialized cache filesystem with storage: {cache_storage}")

Functions

fsspeckit.core.MonitoredSimpleCacheFileSystem.__getattribute__

__getattribute__(item)

Custom attribute access to delegate to underlying filesystem.

This method ensures that attributes not found in this class are looked up in the underlying filesystem.

Source code in src/fsspeckit/core/filesystem/cache.py

def __getattribute__(self, item):
    """Custom attribute access to delegate to underlying filesystem.

    This method ensures that attributes not found in this class
    are looked up in the underlying filesystem.
    """
    if item in {
        # new items
        "size",
        "glob",
        # previous
        "load_cache",
        "_open",
        "save_cache",
        "close_and_update",
        "sync_cache",
        "__init__",
        "__getattribute__",
        "__reduce__",
        "_make_local_details",
        "open",
        "cat",
        "cat_file",
        "cat_ranges",
        "get",
        "read_block",
        "tail",
        "head",
        "info",
        "ls",
        "exists",
        "isfile",
        "isdir",
        "_check_file",
        "_check_cache",
        "_mkcache",
        "clear_cache",
        "clear_expired_cache",
        "pop_from_cache",
        "local_file",
        "_paths_from_path",
        "get_mapper",
        "open_many",
        "commit_many",
        "hash_name",
        "__hash__",
        "__eq__",
        "to_json",
        "to_dict",
        "cache_size",
        "pipe_file",
        "pipe",
        "start_transaction",
        "end_transaction",
        "sync_cache",
    }:
        # all the methods defined in this class. Note `open` here, since
        # it calls `_open`, but is actually in superclass
        return lambda *args, **kw: getattr(type(self), item).__get__(self)(
            *args, **kw
        )
    if item in ["__reduce_ex__"]:
        raise AttributeError
    if item in ["transaction"]:
        # property
        return type(self).transaction.__get__(self)
    if item in ["_cache", "transaction_type"]:
        # class attributes
        return getattr(type(self), item)
    if item == "__class__":
        return type(self)
    d = object.__getattribute__(self, "__dict__")
    fs = d.get("fs", None)  # fs is not immediately defined
    if item in d:
        return d[item]
    elif fs is not None:
        if item in fs.__dict__:
            # attribute of instance
            return fs.__dict__[item]
        # attributed belonging to the target filesystem
        cls = type(fs)
        m = getattr(cls, item)
        if (inspect.isfunction(m) or inspect.ismethod(m)) and (
            not hasattr(m, "__self__") or m.__self__ is None
        ):
            # instance method
            return m.__get__(fs, cls)
        return m  # class method or attribute
    else:
        # attributes of the superclass, while target is being set up
        return super().__getattribute__(item)

fsspeckit.core.MonitoredSimpleCacheFileSystem.open

open(path: str, mode: str = 'rb', **kwargs: Any) -> Any

Open a file from cache or remote filesystem.

Parameters:

Name	Type	Description	Default
path	str	File path	required
mode	str	File mode	'rb'
**kwargs	Any	Additional arguments	{}

Returns:

Type	Description
Any	File-like object

Source code in src/fsspeckit/core/filesystem/cache.py

def open(self, path: str, mode: str = "rb", **kwargs: Any) -> Any:
    """Open a file from cache or remote filesystem.

    Args:
        path: File path
        mode: File mode
        **kwargs: Additional arguments

    Returns:
        File-like object
    """
    return super().open(path, mode=mode, **kwargs)

fsspeckit.core.MonitoredSimpleCacheFileSystem.size

size(path: str) -> int

Get size of file in bytes.

Checks cache first, falls back to remote filesystem.

Parameters:

Name	Type	Description	Default
path	str	Path to file	required

Returns:

Type	Description
int	Size of file in bytes

Example

fs = MonitoredSimpleCacheFileSystem(
    fs=remote_fs,
    cache_storage="/tmp/cache",
)
size = fs.size("large_file.dat")
print(f"File size: {size} bytes")

Source code in src/fsspeckit/core/filesystem/cache.py

def size(self, path: str) -> int:
    """Get size of file in bytes.

    Checks cache first, falls back to remote filesystem.

    Args:
        path: Path to file

    Returns:
        Size of file in bytes

    Example:
        ```python
        fs = MonitoredSimpleCacheFileSystem(
            fs=remote_fs,
            cache_storage="/tmp/cache",
        )
        size = fs.size("large_file.dat")
        print(f"File size: {size} bytes")
        ```
    """
    cached_file = self._check_file(self._strip_protocol(path))
    if cached_file is None:
        return self.fs.size(path)
    else:
        return posixpath.getsize(cached_file)

fsspeckit.core.MonitoredSimpleCacheFileSystem.sync_cache

sync_cache(reload: bool = False) -> None

Synchronize cache with remote filesystem.

Downloads all files in remote path to cache if not present.

Parameters:

Name	Type	Description	Default
reload	bool	Whether to force reload all files, ignoring existing cache	False

Example

fs = MonitoredSimpleCacheFileSystem(
    fs=remote_fs,
    cache_storage="/tmp/cache",
)
# Initial sync
fs.sync_cache()

# Force reload all files
fs.sync_cache(reload=True)

Source code in src/fsspeckit/core/filesystem/cache.py

def sync_cache(self, reload: bool = False) -> None:
    """Synchronize cache with remote filesystem.

    Downloads all files in remote path to cache if not present.

    Args:
        reload: Whether to force reload all files, ignoring existing cache

    Example:
        ```python
        fs = MonitoredSimpleCacheFileSystem(
            fs=remote_fs,
            cache_storage="/tmp/cache",
        )
        # Initial sync
        fs.sync_cache()

        # Force reload all files
        fs.sync_cache(reload=True)
        ```
    """
    if reload:
        if hasattr(self, "clear_cache"):
            self.clear_cache()

    files = self.glob("**/*")
    [self.open(f, mode="rb").close() for f in files if self.isfile(f)]

Functions

fsspeckit.core.get_filesystem

get_filesystem(
    protocol_or_path: Union[str, None] = "",
    storage_options: Union[BaseStorageOptions, dict]
    | None = None,
    **kwargs: Any,
) -> AbstractFileSystem

Get filesystem instance (simple version).

This is a simplified version of filesystem() for backward compatibility. See filesystem() for full documentation.

Parameters:

Name	Type	Description	Default
protocol_or_path	Union[str, None]	Filesystem protocol or path	''
storage_options	Union[BaseStorageOptions, dict] | None	Storage configuration	None
**kwargs	Any	Additional arguments	{}

Returns:

Name	Type	Description
AbstractFileSystem	AbstractFileSystem	Filesystem instance

Source code in src/fsspeckit/core/filesystem/__init__.py

def get_filesystem(
    protocol_or_path: Union[str, None] = "",
    storage_options: Union[BaseStorageOptions, dict] | None = None,
    **kwargs: Any,
) -> AbstractFileSystem:
    """Get filesystem instance (simple version).

    This is a simplified version of filesystem() for backward compatibility.
    See filesystem() for full documentation.

    Args:
        protocol_or_path: Filesystem protocol or path
        storage_options: Storage configuration
        **kwargs: Additional arguments

    Returns:
        AbstractFileSystem: Filesystem instance
    """
    return filesystem(
        protocol_or_path=protocol_or_path,
        storage_options=storage_options,
        **kwargs,
    )