fsspeckit.storage_options API Reference

storage_options

Storage configuration options for different cloud providers and services.

Classes

fsspeckit.storage_options.BaseStorageOptions

Bases: Struct

Base class for filesystem storage configuration options.

Provides common functionality for all storage option classes including: - YAML serialization/deserialization - Dictionary conversion - Filesystem instance creation - Configuration updates

Attributes:

Name	Type	Description
protocol	str	Storage protocol identifier (e.g., "s3", "gs", "file")

Example

# Create and save options
options = BaseStorageOptions(protocol="s3")
options.to_yaml("config.yml")

# Load from YAML
loaded = BaseStorageOptions.from_yaml("config.yml")
print(loaded.protocol)
# 's3'

Functions

fsspeckit.storage_options.BaseStorageOptions.from_yaml classmethod

from_yaml(
    path: str, fs: AbstractFileSystem = None
) -> BaseStorageOptions

Load storage options from YAML file.

Parameters:

Name	Type	Description	Default
path	str	Path to YAML configuration file	required
fs	AbstractFileSystem	Filesystem to use for reading file	None

Returns:

Name	Type	Description
BaseStorageOptions	BaseStorageOptions	Loaded storage options instance

Example

# Load from local file
options = BaseStorageOptions.from_yaml("config.yml")
print(options.protocol)
# 's3'

Source code in src/fsspeckit/storage_options/base.py

@classmethod
def from_yaml(
    cls, path: str, fs: AbstractFileSystem = None
) -> "BaseStorageOptions":
    """Load storage options from YAML file.

    Args:
        path: Path to YAML configuration file
        fs: Filesystem to use for reading file

    Returns:
        BaseStorageOptions: Loaded storage options instance

    Example:
        ```python
        # Load from local file
        options = BaseStorageOptions.from_yaml("config.yml")
        print(options.protocol)
        # 's3'
        ```
    """
    if fs is None:
        fs = fsspec_filesystem("file")
    with fs.open(path) as f:
        data = yaml.safe_load(f)
    return cls(**data)

fsspeckit.storage_options.BaseStorageOptions.to_dict

to_dict(with_protocol: bool = False) -> dict

Convert storage options to dictionary.

Parameters:

Name	Type	Description	Default
with_protocol	bool	Whether to include protocol in output dictionary	False

Returns:

Name	Type	Description
dict	dict	Dictionary of storage options with non-None values

Example

options = BaseStorageOptions(protocol="s3")
print(options.to_dict())
# {}
print(options.to_dict(with_protocol=True))
# {'protocol': 's3'}

Source code in src/fsspeckit/storage_options/base.py

def to_dict(self, with_protocol: bool = False) -> dict:
    """Convert storage options to dictionary.

    Args:
        with_protocol: Whether to include protocol in output dictionary

    Returns:
        dict: Dictionary of storage options with non-None values

    Example:
        ```python
        options = BaseStorageOptions(protocol="s3")
        print(options.to_dict())
        # {}
        print(options.to_dict(with_protocol=True))
        # {'protocol': 's3'}
        ```
    """
    data = msgspec.structs.asdict(self)
    result = {}
    for key, value in data.items():
        if value is None:
            continue

        if key == "protocol":
            if with_protocol:
                result[key] = value
        else:
            result[key] = value
    return result

fsspeckit.storage_options.BaseStorageOptions.to_filesystem

to_filesystem(
    use_listings_cache: bool = True,
    skip_instance_cache: bool = False,
    **kwargs: Any,
) -> AbstractFileSystem

Create fsspec filesystem instance from options.

Parameters:

Name	Type	Description	Default
use_listings_cache	bool	Whether to enable the fsspec listings cache.	True
skip_instance_cache	bool	Whether to skip fsspec's instance cache.	False
**kwargs	Any	Additional arguments forwarded to fsspec.filesystem.	{}

Returns:

Name	Type	Description
AbstractFileSystem	AbstractFileSystem	Configured filesystem instance

Example

options = BaseStorageOptions(protocol="file")
fs = options.to_filesystem()
files = fs.ls("/path/to/data")

Source code in src/fsspeckit/storage_options/base.py

def to_filesystem(
    self,
    use_listings_cache: bool = True,
    skip_instance_cache: bool = False,
    **kwargs: Any,
) -> AbstractFileSystem:
    """Create fsspec filesystem instance from options.

    Args:
        use_listings_cache: Whether to enable the fsspec listings cache.
        skip_instance_cache: Whether to skip fsspec's instance cache.
        **kwargs: Additional arguments forwarded to ``fsspec.filesystem``.

    Returns:
        AbstractFileSystem: Configured filesystem instance

    Example:
        ```python
        options = BaseStorageOptions(protocol="file")
        fs = options.to_filesystem()
        files = fs.ls("/path/to/data")
        ```
    """

    fsspec_kwargs: dict[str, Any]
    if hasattr(self, "to_fsspec_kwargs"):
        to_kwargs = getattr(self, "to_fsspec_kwargs")
        try:
            fsspec_kwargs = to_kwargs(
                use_listings_cache=use_listings_cache,
                skip_instance_cache=skip_instance_cache,
            )
        except TypeError:
            fsspec_kwargs = to_kwargs()
    else:
        fsspec_kwargs = self.to_dict(with_protocol=False)

    merged_kwargs: dict[str, Any] = {}
    if fsspec_kwargs:
        merged_kwargs.update(fsspec_kwargs)

    merged_kwargs.setdefault("use_listings_cache", use_listings_cache)
    merged_kwargs.setdefault("skip_instance_cache", skip_instance_cache)
    merged_kwargs.update({k: v for k, v in kwargs.items() if v is not None})

    filtered_kwargs = {k: v for k, v in merged_kwargs.items() if v is not None}

    return fsspec_filesystem(self.protocol, **filtered_kwargs)

fsspeckit.storage_options.BaseStorageOptions.to_yaml

to_yaml(path: str, fs: AbstractFileSystem = None) -> None

Save storage options to YAML file.

Parameters:

Name	Type	Description	Default
path	str	Path where to save configuration	required
fs	AbstractFileSystem	Filesystem to use for writing	None

Example

options = BaseStorageOptions(protocol="s3")
options.to_yaml("config.yml")

Source code in src/fsspeckit/storage_options/base.py

def to_yaml(self, path: str, fs: AbstractFileSystem = None) -> None:
    """Save storage options to YAML file.

    Args:
        path: Path where to save configuration
        fs: Filesystem to use for writing

    Example:
        ```python
        options = BaseStorageOptions(protocol="s3")
        options.to_yaml("config.yml")
        ```
    """
    if fs is None:
        fs = fsspec_filesystem("file")
    data = self.to_dict()
    with fs.open(path, "w") as f:
        yaml.safe_dump(data, f)

fsspeckit.storage_options.BaseStorageOptions.update

update(**kwargs: Any) -> BaseStorageOptions

Update storage options with new values.

Parameters:

Name	Type	Description	Default
**kwargs	Any	New option values to set	{}

Returns:

Name	Type	Description
BaseStorageOptions	BaseStorageOptions	Updated instance

Example

options = BaseStorageOptions(protocol="s3")
options = options.update(region="us-east-1")
print(options.region)
# 'us-east-1'

Source code in src/fsspeckit/storage_options/base.py

def update(self, **kwargs: Any) -> "BaseStorageOptions":
    """Update storage options with new values.

    Args:
        **kwargs: New option values to set

    Returns:
        BaseStorageOptions: Updated instance

    Example:
        ```python
        options = BaseStorageOptions(protocol="s3")
        options = options.update(region="us-east-1")
        print(options.region)
        # 'us-east-1'
        ```
    """
    return msgspec.structs.replace(self, **kwargs)

fsspeckit.storage_options.StorageOptions

Bases: Struct

High-level storage options container and factory.

Provides a unified interface for creating and managing storage options for different protocols.

Attributes:

Name	Type	Description
storage_options	BaseStorageOptions	Underlying storage options instance

Example

# Create from protocol
options = StorageOptions.create(
    protocol="s3",
    access_key_id="KEY",
    secret_access_key="SECRET",
)

# Create from existing options
s3_opts = AwsStorageOptions(access_key_id="KEY")
options = StorageOptions(storage_options=s3_opts)

Functions

fsspeckit.storage_options.StorageOptions.create classmethod

create(**data: Any) -> StorageOptions

Create storage options from arguments.

Parameters:

Name	Type	Description	Default
**data	Any	Either: - protocol and configuration options - storage_options=pre-configured instance	{}

Returns:

Name	Type	Description
StorageOptions	StorageOptions	Configured storage options instance

Raises:

Type	Description
ValueError	If protocol missing or invalid

Example

Direct protocol config

options = StorageOptions.create( ... protocol="s3", ... region="us-east-1" ... )

Source code in src/fsspeckit/storage_options/core.py

@classmethod
def create(cls, **data: Any) -> "StorageOptions":
    """Create storage options from arguments.

    Args:
        **data: Either:
            - protocol and configuration options
            - storage_options=pre-configured instance

    Returns:
        StorageOptions: Configured storage options instance

    Raises:
        ValueError: If protocol missing or invalid

    Example:
        >>> # Direct protocol config
        >>> options = StorageOptions.create(
        ...     protocol="s3",
        ...     region="us-east-1"
        ... )
    """
    protocol = data.get("protocol")
    if protocol is None and "storage_options" not in data:
        raise ValueError("protocol must be specified")

    if "storage_options" not in data:
        if protocol == "s3":
            if "profile" in data or "key" in data or "secret" in data:
                storage_options = AwsStorageOptions.create(**data)
            else:
                storage_options = AwsStorageOptions(**data)
        elif protocol == "github":
            storage_options = GitHubStorageOptions(**data)
        elif protocol == "gitlab":
            storage_options = GitLabStorageOptions(**data)
        elif protocol in ["az", "abfs", "adl"]:
            storage_options = AzureStorageOptions(**data)
        elif protocol in ["gs", "gcs"]:
            storage_options = GcsStorageOptions(**data)
        elif protocol == "file":
            storage_options = LocalStorageOptions(**data)
        else:
            raise ValueError(f"Unsupported protocol: {protocol}")

        return cls(storage_options=storage_options)
    else:
        return cls(**data)

fsspeckit.storage_options.StorageOptions.from_env classmethod

from_env(protocol: str) -> StorageOptions

Create storage options from environment variables.

Parameters:

Name	Type	Description	Default
protocol	str	Storage protocol to configure	required

Returns:

Name	Type	Description
StorageOptions	StorageOptions	Environment-configured options

Example

Load AWS config from environment

options = StorageOptions.from_env("s3")

Source code in src/fsspeckit/storage_options/core.py

@classmethod
def from_env(cls, protocol: str) -> "StorageOptions":
    """Create storage options from environment variables.

    Args:
        protocol: Storage protocol to configure

    Returns:
        StorageOptions: Environment-configured options

    Example:
        >>> # Load AWS config from environment
        >>> options = StorageOptions.from_env("s3")
    """
    storage_options = from_env(protocol)
    return cls(storage_options=storage_options)

fsspeckit.storage_options.StorageOptions.from_yaml classmethod

from_yaml(
    path: str, fs: AbstractFileSystem = None
) -> StorageOptions

Create storage options from YAML configuration.

Parameters:

Name	Type	Description	Default
path	str	Path to YAML configuration file	required
fs	AbstractFileSystem	Filesystem for reading configuration	None

Returns:

Name	Type	Description
StorageOptions	StorageOptions	Configured storage options

Example

Load from config file

options = StorageOptions.from_yaml("storage.yml") print(options.storage_options.protocol) 's3'

Source code in src/fsspeckit/storage_options/core.py

@classmethod
def from_yaml(cls, path: str, fs: AbstractFileSystem = None) -> "StorageOptions":
    """Create storage options from YAML configuration.

    Args:
        path: Path to YAML configuration file
        fs: Filesystem for reading configuration

    Returns:
        StorageOptions: Configured storage options

    Example:
        >>> # Load from config file
        >>> options = StorageOptions.from_yaml("storage.yml")
        >>> print(options.storage_options.protocol)
        's3'
    """
    if fs is None:
        fs = fsspec_filesystem("file")
    with fs.open(path, "r") as f:
        data = yaml.safe_load(f)
    return cls.create(**data)

fsspeckit.storage_options.StorageOptions.to_dict

to_dict(with_protocol: bool = False) -> dict

Convert storage options to dictionary.

Parameters:

Name	Type	Description	Default
with_protocol	bool	Whether to include protocol in output	False

Returns:

Name	Type	Description
dict	dict	Storage options as dictionary

Example

options = StorageOptions.create( ... protocol="s3", ... region="us-east-1" ... ) print(options.to_dict())

Source code in src/fsspeckit/storage_options/core.py

def to_dict(self, with_protocol: bool = False) -> dict:
    """Convert storage options to dictionary.

    Args:
        with_protocol: Whether to include protocol in output

    Returns:
        dict: Storage options as dictionary

    Example:
        >>> options = StorageOptions.create(
        ...     protocol="s3",
        ...     region="us-east-1"
        ... )
        >>> print(options.to_dict())
        {'region': 'us-east-1'}
    """
    return self.storage_options.to_dict(with_protocol=with_protocol)

fsspeckit.storage_options.StorageOptions.to_filesystem

to_filesystem(
    use_listings_cache: bool = True,
    skip_instance_cache: bool = False,
    **kwargs: Any,
) -> AbstractFileSystem

Create fsspec filesystem instance.

Returns:

Name	Type	Description
AbstractFileSystem	AbstractFileSystem	Configured filesystem instance

Example

options = StorageOptions.create(protocol="file") fs = options.to_filesystem() files = fs.ls("/data")

Source code in src/fsspeckit/storage_options/core.py

def to_filesystem(
    self,
    use_listings_cache: bool = True,
    skip_instance_cache: bool = False,
    **kwargs: Any,
) -> AbstractFileSystem:
    """Create fsspec filesystem instance.

    Returns:
        AbstractFileSystem: Configured filesystem instance

    Example:
        >>> options = StorageOptions.create(protocol="file")
        >>> fs = options.to_filesystem()
        >>> files = fs.ls("/data")
    """
    return self.storage_options.to_filesystem(
        use_listings_cache=use_listings_cache,
        skip_instance_cache=skip_instance_cache,
        **kwargs,
    )

fsspeckit.storage_options.StorageOptions.to_object_store_kwargs

to_object_store_kwargs(
    with_conditional_put: bool = False,
) -> dict

Get options formatted for object store clients.

Parameters:

Name	Type	Description	Default
with_conditional_put	bool	Add etag-based conditional put support	False

Returns:

Name	Type	Description
dict	dict	Object store configuration dictionary

Example

options = StorageOptions.create(protocol="s3") kwargs = options.to_object_store_kwargs()

store = ObjectStore(**kwargs)

Source code in src/fsspeckit/storage_options/core.py

def to_object_store_kwargs(self, with_conditional_put: bool = False) -> dict:
    """Get options formatted for object store clients.

    Args:
        with_conditional_put: Add etag-based conditional put support

    Returns:
        dict: Object store configuration dictionary

    Example:
        >>> options = StorageOptions.create(protocol="s3")
        >>> kwargs = options.to_object_store_kwargs()
        >>> # store = ObjectStore(**kwargs)
    """
    if hasattr(self.storage_options, "to_object_store_kwargs"):
        return self.storage_options.to_object_store_kwargs(
            with_conditional_put=with_conditional_put
        )
    else:
        return self.storage_options.to_dict(with_protocol=True)

fsspeckit.storage_options.LocalStorageOptions

Bases: BaseStorageOptions

Local filesystem configuration options.

Provides basic configuration for local file access. While this class is simple, it maintains consistency with other storage options and enables transparent switching between local and remote storage.

Attributes:

Name	Type	Description
protocol	str	Always "file" for local filesystem
auto_mkdir	bool	Create directories automatically
mode	int	Default file creation mode (unix-style)

Example

# Basic local access
options = LocalStorageOptions()
fs = options.to_filesystem()
files = fs.ls("/path/to/data")

# With auto directory creation
options = LocalStorageOptions(auto_mkdir=True)
fs = options.to_filesystem()
with fs.open("/new/path/file.txt", "w") as f:
    f.write("test")  # Creates /new/path/ automatically

Functions

fsspeckit.storage_options.LocalStorageOptions.to_fsspec_kwargs

to_fsspec_kwargs() -> dict

Convert options to fsspec filesystem arguments.

Returns:

Name	Type	Description
dict	dict	Arguments suitable for LocalFileSystem

Example

options = LocalStorageOptions(auto_mkdir=True)
kwargs = options.to_fsspec_kwargs()
fs = filesystem("file", **kwargs)

Source code in src/fsspeckit/storage_options/core.py

def to_fsspec_kwargs(self) -> dict:
    """Convert options to fsspec filesystem arguments.

    Returns:
        dict: Arguments suitable for LocalFileSystem

    Example:
        ```python
        options = LocalStorageOptions(auto_mkdir=True)
        kwargs = options.to_fsspec_kwargs()
        fs = filesystem("file", **kwargs)
        ```
    """
    kwargs = {
        "auto_mkdir": self.auto_mkdir,
        "mode": self.mode,
    }
    return {k: v for k, v in kwargs.items() if v is not None}

fsspeckit.storage_options.AwsStorageOptions

Bases: BaseStorageOptions

AWS S3 storage configuration options.

Provides comprehensive configuration for S3 access with support for: - Multiple authentication methods (keys, profiles, environment) - Custom endpoints for S3-compatible services - Region configuration - SSL/TLS settings - Anonymous access for public buckets

Attributes:

Name	Type	Description
protocol	str	Always "s3" for S3 storage
access_key_id	str	AWS access key ID
secret_access_key	str	AWS secret access key
session_token	str	AWS session token
endpoint_url	str	Custom S3 endpoint URL
region	str	AWS region name
allow_invalid_certificates	bool	Skip SSL certificate validation
allow_http	bool	Allow unencrypted HTTP connections
anonymous	bool	Use anonymous (unsigned) S3 access

Example

Basic credentials

options = AwsStorageOptions( ... access_key_id="AKIAXXXXXXXX", ... secret_access_key="SECRETKEY", ... region="us-east-1" ... )

Profile-based auth

options = AwsStorageOptions.create(profile="dev")

S3-compatible service (MinIO)

options = AwsStorageOptions( ... endpoint_url="http://localhost:9000", ... access_key_id="minioadmin", ... secret_access_key="minioadmin", ... allow_http=True ... )

Anonymous access for public buckets

options = AwsStorageOptions(anonymous=True)

Functions

fsspeckit.storage_options.AwsStorageOptions.create classmethod

create(
    protocol: str = "s3",
    access_key_id: str | None = None,
    secret_access_key: str | None = None,
    session_token: str | None = None,
    endpoint_url: str | None = None,
    region: str | None = None,
    allow_invalid_certificates: bool | str | None = None,
    allow_invalid_certs: bool | str | None = None,
    allow_http: bool | str | None = None,
    anonymous: bool | str | None = None,
    key: str | None = None,
    secret: str | None = None,
    token: str | None = None,
    profile: str | None = None,
) -> AwsStorageOptions

Creates an AwsStorageOptions instance, handling aliases and profile loading.

Parameters:

Name	Type	Description	Default
protocol	str	Storage protocol, defaults to "s3".	's3'
access_key_id	str | None	AWS access key ID.	None
secret_access_key	str | None	AWS secret access key.	None
session_token	str | None	AWS session token.	None
endpoint_url	str | None	Custom S3 endpoint URL.	None
region	str | None	AWS region name.	None
allow_invalid_certificates	bool | str | None	Skip SSL certificate validation.	None
allow_http	bool | str | None	Allow unencrypted HTTP connections.	None
anonymous	bool | str | None	Use anonymous (unsigned) S3 access.	None
key	str | None	Alias for access_key_id.	None
secret	str | None	Alias for secret_access_key.	None
token	str | None	Alias for session_token.	None
profile	str | None	AWS credentials profile name to load credentials from.	None

Returns:

Type	Description
AwsStorageOptions	An initialized AwsStorageOptions instance.

Source code in src/fsspeckit/storage_options/cloud.py

@classmethod
def create(
    cls,
    protocol: str = "s3",
    access_key_id: str | None = None,
    secret_access_key: str | None = None,
    session_token: str | None = None,
    endpoint_url: str | None = None,
    region: str | None = None,
    allow_invalid_certificates: bool | str | None = None,
    allow_invalid_certs: bool | str | None = None,
    allow_http: bool | str | None = None,
    anonymous: bool | str | None = None,
    # Alias and loading params
    key: str | None = None,
    secret: str | None = None,
    token: str | None = None,  # maps to session_token
    profile: str | None = None,
) -> "AwsStorageOptions":
    """Creates an AwsStorageOptions instance, handling aliases and profile loading.

    Args:
        protocol: Storage protocol, defaults to "s3".
        access_key_id: AWS access key ID.
        secret_access_key: AWS secret access key.
        session_token: AWS session token.
        endpoint_url: Custom S3 endpoint URL.
        region: AWS region name.
        allow_invalid_certificates: Skip SSL certificate validation.
        allow_http: Allow unencrypted HTTP connections.
        anonymous: Use anonymous (unsigned) S3 access.
        key: Alias for access_key_id.
        secret: Alias for secret_access_key.
        token: Alias for session_token.
        profile: AWS credentials profile name to load credentials from.

    Returns:
        An initialized AwsStorageOptions instance.
    """
    protocol = protocol or "s3"
    canonical_allow_invalid = _parse_bool(allow_invalid_certificates)
    alias_allow_invalid = _parse_bool(allow_invalid_certs)
    if canonical_allow_invalid is None:
        canonical_allow_invalid = alias_allow_invalid

    allow_http_flag = _parse_bool(allow_http)
    anonymous_flag = _parse_bool(anonymous)

    # Initial values from explicit args or their aliases
    args = {
        "protocol": protocol,
        "access_key_id": access_key_id if access_key_id is not None else key,
        "secret_access_key": secret_access_key
        if secret_access_key is not None
        else secret,
        "session_token": session_token if session_token is not None else token,
        "endpoint_url": endpoint_url,
        "region": region,
        "allow_invalid_certificates": canonical_allow_invalid,
        "allow_invalid_certs": None,
        "allow_http": allow_http_flag,
        "anonymous": anonymous_flag,
    }

    if profile is not None:
        profile_instance = cls.from_aws_credentials(
            profile=profile,
            allow_invalid_certificates=args["allow_invalid_certificates"],
            allow_http=args["allow_http"],
            anonymous=args["anonymous"],
        )
        # Fill in missing values from profile if not already set by direct/aliased args
        if args["access_key_id"] is None:
            args["access_key_id"] = profile_instance.access_key_id
        if args["secret_access_key"] is None:
            args["secret_access_key"] = profile_instance.secret_access_key
        if args["session_token"] is None:
            args["session_token"] = profile_instance.session_token
        if args["endpoint_url"] is None:
            args["endpoint_url"] = profile_instance.endpoint_url
        if args["region"] is None:
            args["region"] = profile_instance.region
        if (
            args["allow_invalid_certificates"] is None
            and profile_instance.allow_invalid_certificates is not None
        ):
            args["allow_invalid_certificates"] = (
                profile_instance.allow_invalid_certificates
            )
        if args["allow_http"] is None and profile_instance.allow_http is not None:
            args["allow_http"] = profile_instance.allow_http

    # Ensure protocol is 's3' if it somehow became None
    if args["protocol"] is None:
        args["protocol"] = "s3"

    return cls(**args)  # type: ignore[arg-type]

fsspeckit.storage_options.AwsStorageOptions.from_aws_credentials classmethod

from_aws_credentials(
    profile: str,
    allow_invalid_certificates: bool | str | None = None,
    allow_invalid_certs: bool | str | None = None,
    allow_http: bool | str | None = None,
    anonymous: bool | str | None = None,
) -> AwsStorageOptions

Create storage options from AWS credentials file.

Loads credentials from ~/.aws/credentials and ~/.aws/config files.

Parameters:

Name	Type	Description	Default
profile	str	AWS credentials profile name	required
allow_invalid_certificates	bool | str | None	Skip SSL certificate validation	None
allow_invalid_certs	bool | str | None	Skip SSL certificate validation (deprecated, use allow_invalid_certificates)	None
allow_http	bool | str | None	Allow unencrypted HTTP connections	None
anonymous	bool | str | None	Use anonymous (unsigned) S3 access	None

Returns:

Name	Type	Description
AwsStorageOptions	AwsStorageOptions	Configured storage options

Raises:

Type	Description
ValueError	If profile not found
FileNotFoundError	If credentials files missing

Example

Load developer profile

options = AwsStorageOptions.from_aws_credentials( ... profile="dev", ... allow_http=True # For local testing ... )

Source code in src/fsspeckit/storage_options/cloud.py

@classmethod
def from_aws_credentials(
    cls,
    profile: str,
    allow_invalid_certificates: bool | str | None = None,
    allow_invalid_certs: bool | str | None = None,
    allow_http: bool | str | None = None,
    anonymous: bool | str | None = None,
) -> "AwsStorageOptions":
    """Create storage options from AWS credentials file.

    Loads credentials from ~/.aws/credentials and ~/.aws/config files.

    Args:
        profile: AWS credentials profile name
        allow_invalid_certificates: Skip SSL certificate validation
        allow_invalid_certs: Skip SSL certificate validation (deprecated, use allow_invalid_certificates)
        allow_http: Allow unencrypted HTTP connections
        anonymous: Use anonymous (unsigned) S3 access

    Returns:
        AwsStorageOptions: Configured storage options

    Raises:
        ValueError: If profile not found
        FileNotFoundError: If credentials files missing

    Example:
        >>> # Load developer profile
        >>> options = AwsStorageOptions.from_aws_credentials(
        ...     profile="dev",
        ...     allow_http=True  # For local testing
        ... )
    """
    cp = configparser.ConfigParser()
    cp.read(os.path.expanduser("~/.aws/credentials"))
    cp.read(os.path.expanduser("~/.aws/config"))
    if profile not in cp:
        raise ValueError(f"Profile '{profile}' not found in AWS credentials file")

    canonical_allow_invalid = _parse_bool(allow_invalid_certificates)
    alias_allow_invalid = _parse_bool(allow_invalid_certs)
    if canonical_allow_invalid is None:
        canonical_allow_invalid = alias_allow_invalid

    allow_http_flag = _parse_bool(allow_http)
    anonymous_flag = _parse_bool(anonymous)

    return cls(
        protocol="s3",
        access_key_id=cp[profile].get("aws_access_key_id", None),
        secret_access_key=cp[profile].get("aws_secret_access_key", None),
        session_token=cp[profile].get("aws_session_token", None),
        endpoint_url=cp[profile].get("aws_endpoint_url", None)
        or cp[profile].get("endpoint_url", None)
        or cp[profile].get("aws_endpoint", None)
        or cp[profile].get("endpoint", None),
        region=(
            cp[profile].get("region", None)
            or cp[f"profile {profile}"].get("region", None)
            if f"profile {profile}" in cp
            else None
        ),
        allow_invalid_certificates=canonical_allow_invalid,
        allow_invalid_certs=None,
        allow_http=allow_http_flag,
        anonymous=anonymous_flag,
    )

fsspeckit.storage_options.AwsStorageOptions.from_env classmethod

from_env() -> AwsStorageOptions

Create storage options from environment variables.

Reads standard AWS environment variables:
- AWS_ACCESS_KEY_ID
- AWS_SECRET_ACCESS_KEY
- AWS_SESSION_TOKEN
- AWS_ENDPOINT_URL
- AWS_DEFAULT_REGION
- ALLOW_INVALID_CERTIFICATES
- AWS_ALLOW_HTTP
- AWS_S3_ANONYMOUS

Returns:

Name	Type	Description
AwsStorageOptions	AwsStorageOptions	Configured storage options

Example:
    ```python
    # Load from environment
    options = AwsStorageOptions.from_env()
    print(options.region)
    # 'us-east-1'  # From AWS_DEFAULT_REGION
    ```

Source code in src/fsspeckit/storage_options/cloud.py

@classmethod
def from_env(cls) -> "AwsStorageOptions":
    """Create storage options from environment variables.

    Reads standard AWS environment variables:
    - AWS_ACCESS_KEY_ID
    - AWS_SECRET_ACCESS_KEY
    - AWS_SESSION_TOKEN
    - AWS_ENDPOINT_URL
    - AWS_DEFAULT_REGION
    - ALLOW_INVALID_CERTIFICATES
    - AWS_ALLOW_HTTP
    - AWS_S3_ANONYMOUS

Returns:
        AwsStorageOptions: Configured storage options

    Example:
        ```python
        # Load from environment
        options = AwsStorageOptions.from_env()
        print(options.region)
        # 'us-east-1'  # From AWS_DEFAULT_REGION
        ```
    """
    return cls(
        access_key_id=os.getenv("AWS_ACCESS_KEY_ID"),
        secret_access_key=os.getenv("AWS_SECRET_ACCESS_KEY"),
        session_token=os.getenv("AWS_SESSION_TOKEN"),
        endpoint_url=os.getenv("AWS_ENDPOINT_URL"),
        region=os.getenv("AWS_DEFAULT_REGION"),
        allow_invalid_certificates=_parse_bool(
            os.getenv("ALLOW_INVALID_CERTIFICATES")
        ),
        allow_invalid_certs=None,
        allow_http=_parse_bool(os.getenv("AWS_ALLOW_HTTP")),
        anonymous=_parse_bool(os.getenv("AWS_S3_ANONYMOUS")),
    )

fsspeckit.storage_options.AwsStorageOptions.to_env

to_env() -> None

Export options to environment variables.

Sets standard AWS environment variables.

Example

options = AwsStorageOptions(
    access_key_id="KEY",
    secret_access_key="SECRET",
    region="us-east-1",
)
options.to_env()
print(os.getenv("AWS_ACCESS_KEY_ID"))
# 'KEY'

Source code in src/fsspeckit/storage_options/cloud.py

def to_env(self) -> None:
    """Export options to environment variables.

    Sets standard AWS environment variables.

    Example:
        ```python
        options = AwsStorageOptions(
            access_key_id="KEY",
            secret_access_key="SECRET",
            region="us-east-1",
        )
        options.to_env()
        print(os.getenv("AWS_ACCESS_KEY_ID"))
        # 'KEY'
        ```
    """
    env = {
        "AWS_ACCESS_KEY_ID": self.access_key_id,
        "AWS_SECRET_ACCESS_KEY": self.secret_access_key,
        "AWS_SESSION_TOKEN": self.session_token,
        "AWS_ENDPOINT_URL": self.endpoint_url,
        "AWS_DEFAULT_REGION": self.region,
        "ALLOW_INVALID_CERTIFICATES": self._parsed_allow_invalid_certificates,
        "AWS_ALLOW_HTTP": self._parsed_allow_http,
        "AWS_S3_ANONYMOUS": self._parsed_anonymous,
    }
    env = {k: str(v) for k, v in env.items() if v is not None}
    os.environ.update(env)  # type: ignore[arg-type]

fsspeckit.storage_options.AwsStorageOptions.to_fsspec_kwargs

to_fsspec_kwargs(
    use_listings_cache: bool = True,
    skip_instance_cache: bool = False,
) -> dict

Convert options to fsspec filesystem arguments. Args: conditional_put: Enable etag-based conditional puts use_listings_cache: Enable fsspec listings cache skip_instance_cache: Skip fsspec instance cache

Returns:

Name	Type	Description
dict	dict	Arguments suitable for fsspec S3FileSystem

Example

options = AwsStorageOptions(
    access_key_id="KEY",
    secret_access_key="SECRET",
    region="us-west-2",
)
kwargs = options.to_fsspec_kwargs()
fs = filesystem("s3", **kwargs)

Source code in src/fsspeckit/storage_options/cloud.py

def to_fsspec_kwargs(
    self,
    use_listings_cache: bool = True,
    skip_instance_cache: bool = False,
) -> dict:
    """Convert options to fsspec filesystem arguments.
    Args:
        conditional_put: Enable etag-based conditional puts
        use_listings_cache: Enable fsspec listings cache
        skip_instance_cache: Skip fsspec instance cache

    Returns:
        dict: Arguments suitable for fsspec S3FileSystem

    Example:
        ```python
        options = AwsStorageOptions(
            access_key_id="KEY",
            secret_access_key="SECRET",
            region="us-west-2",
        )
        kwargs = options.to_fsspec_kwargs()
        fs = filesystem("s3", **kwargs)
        ```
    """
    fsspec_kwargs: dict[str, Any] = {
        "endpoint_url": self.endpoint_url,
        "use_listings_cache": use_listings_cache,
        "skip_instance_cache": skip_instance_cache,
    }

    # Handle anonymous access
    if self._parsed_anonymous:
        fsspec_kwargs["anon"] = True
    else:
        # Include credentials only if not anonymous
        fsspec_kwargs.update(
            {
                "key": self.access_key_id,
                "secret": self.secret_access_key,
                "token": self.session_token,
            }
        )

    verify_value = (
        None
        if self._parsed_allow_invalid_certificates is None
        else not self._parsed_allow_invalid_certificates
    )
    use_ssl_value = (
        None if self._parsed_allow_http is None else not self._parsed_allow_http
    )

    client_kwargs = {
        "region_name": self.region,
        "verify": verify_value,
        "use_ssl": use_ssl_value,
    }
    client_kwargs = {k: v for k, v in client_kwargs.items() if v is not None}
    if client_kwargs:
        fsspec_kwargs["client_kwargs"] = client_kwargs

    return {k: v for k, v in fsspec_kwargs.items() if v is not None}

fsspeckit.storage_options.AwsStorageOptions.to_object_store_kwargs

to_object_store_kwargs(
    *,
    with_conditional_put: bool | str = False,
    conditional_put: str | None = None,
) -> dict

Convert options to object store arguments.

Source code in src/fsspeckit/storage_options/cloud.py

def to_object_store_kwargs(
    self,
    *,
    with_conditional_put: bool | str = False,
    conditional_put: str | None = None,
) -> dict:
    """Convert options to object store arguments."""

    storage_options = {
        k: str(v)
        for k, v in self.to_dict().items()
        if v is not None and k != "protocol"
    }

    cond_value: str | None = None
    if conditional_put is not None:
        cond_value = conditional_put
    elif isinstance(with_conditional_put, str):
        cond_value = with_conditional_put
    elif with_conditional_put:
        cond_value = "etag"

    if cond_value:
        storage_options["conditional_put"] = cond_value

    return storage_options

fsspeckit.storage_options.GcsStorageOptions

Bases: BaseStorageOptions

Google Cloud Storage configuration options.

Provides configuration for GCS access with support for: - Service account authentication - Default application credentials - Token-based authentication - Project configuration - Custom endpoints

Attributes:

Name	Type	Description
protocol	str	Storage protocol ("gs" or "gcs")
token	str	Path to service account JSON file
project	str	Google Cloud project ID
access_token	str	OAuth2 access token
endpoint_url	str	Custom storage endpoint
timeout	int	Request timeout in seconds

Example

Service account auth

options = GcsStorageOptions( ... protocol="gs", ... token="path/to/service-account.json", ... project="my-project-123" ... )

Application default credentials

options = GcsStorageOptions( ... protocol="gcs", ... project="my-project-123" ... )

Custom endpoint (e.g., test server)

options = GcsStorageOptions( ... protocol="gs", ... endpoint_url="http://localhost:4443", ... token="test-token.json" ... )

Functions

fsspeckit.storage_options.GcsStorageOptions.from_env classmethod

from_env() -> GcsStorageOptions

Create storage options from environment variables.

Reads standard GCP environment variables: - GOOGLE_CLOUD_PROJECT: Project ID - GOOGLE_APPLICATION_CREDENTIALS: Service account file path - STORAGE_EMULATOR_HOST: Custom endpoint (for testing) - GCS_OAUTH_TOKEN: OAuth2 access token

Returns:

Name	Type	Description
GcsStorageOptions	GcsStorageOptions	Configured storage options

Example

With environment variables set:

options = GcsStorageOptions.from_env() print(options.project) # From GOOGLE_CLOUD_PROJECT 'my-project-123'

Source code in src/fsspeckit/storage_options/cloud.py

@classmethod
def from_env(cls) -> "GcsStorageOptions":
    """Create storage options from environment variables.

    Reads standard GCP environment variables:
    - GOOGLE_CLOUD_PROJECT: Project ID
    - GOOGLE_APPLICATION_CREDENTIALS: Service account file path
    - STORAGE_EMULATOR_HOST: Custom endpoint (for testing)
    - GCS_OAUTH_TOKEN: OAuth2 access token

    Returns:
        GcsStorageOptions: Configured storage options

    Example:
        >>> # With environment variables set:
        >>> options = GcsStorageOptions.from_env()
        >>> print(options.project)  # From GOOGLE_CLOUD_PROJECT
        'my-project-123'
    """
    return cls(
        protocol="gs",
        project=os.getenv("GOOGLE_CLOUD_PROJECT"),
        token=os.getenv("GOOGLE_APPLICATION_CREDENTIALS"),
        endpoint_url=os.getenv("STORAGE_EMULATOR_HOST"),
        access_token=os.getenv("GCS_OAUTH_TOKEN"),
    )

fsspeckit.storage_options.GcsStorageOptions.to_env

to_env() -> None

Export options to environment variables.

Sets standard GCP environment variables.

Example

options = GcsStorageOptions( ... protocol="gs", ... project="my-project", ... token="service-account.json" ... ) options.to_env() print(os.getenv("GOOGLE_CLOUD_PROJECT")) 'my-project'

Source code in src/fsspeckit/storage_options/cloud.py

def to_env(self) -> None:
    """Export options to environment variables.

    Sets standard GCP environment variables.

    Example:
        >>> options = GcsStorageOptions(
        ...     protocol="gs",
        ...     project="my-project",
        ...     token="service-account.json"
        ... )
        >>> options.to_env()
        >>> print(os.getenv("GOOGLE_CLOUD_PROJECT"))
        'my-project'
    """
    env = {
        "GOOGLE_CLOUD_PROJECT": self.project,
        "GOOGLE_APPLICATION_CREDENTIALS": self.token,
        "STORAGE_EMULATOR_HOST": self.endpoint_url,
        "GCS_OAUTH_TOKEN": self.access_token,
    }
    env = {k: str(v) for k, v in env.items() if v is not None}
    os.environ.update(env)  # type: ignore[arg-type]

fsspeckit.storage_options.GcsStorageOptions.to_fsspec_kwargs

to_fsspec_kwargs() -> dict

Convert options to fsspec filesystem arguments.

Returns:

Name	Type	Description
dict	dict	Arguments suitable for GCSFileSystem

Example

options = GcsStorageOptions( ... protocol="gs", ... token="service-account.json", ... project="my-project" ... ) kwargs = options.to_fsspec_kwargs() fs = filesystem("gcs", **kwargs)

Source code in src/fsspeckit/storage_options/cloud.py

def to_fsspec_kwargs(self) -> dict:
    """Convert options to fsspec filesystem arguments.

    Returns:
        dict: Arguments suitable for GCSFileSystem

    Example:
        >>> options = GcsStorageOptions(
        ...     protocol="gs",
        ...     token="service-account.json",
        ...     project="my-project"
        ... )
        >>> kwargs = options.to_fsspec_kwargs()
        >>> fs = filesystem("gcs", **kwargs)
    """
    kwargs = {
        "token": self.token,
        "project": self.project,
        "access_token": self.access_token,
        "endpoint_url": self.endpoint_url,
        "timeout": self.timeout,
    }
    return {k: v for k, v in kwargs.items() if v is not None}

fsspeckit.storage_options.AzureStorageOptions

Bases: BaseStorageOptions

Azure Storage configuration options.

Provides configuration for Azure storage services: - Azure Blob Storage (az://) - Azure Data Lake Storage Gen2 (abfs://) - Azure Data Lake Storage Gen1 (adl://)

Supports multiple authentication methods: - Connection string - Account key - Service principal - Managed identity - SAS token

Attributes:

Name	Type	Description
protocol	str	Storage protocol ("az", "abfs", or "adl")
account_name	str	Storage account name
account_key	str	Storage account access key
connection_string	str	Full connection string
tenant_id	str	Azure AD tenant ID
client_id	str	Service principal client ID
client_secret	str	Service principal client secret
sas_token	str	SAS token for limited access

Example

Blob Storage with account key

options = AzureStorageOptions( ... protocol="az", ... account_name="mystorageacct", ... account_key="key123..." ... )

Data Lake with service principal

options = AzureStorageOptions( ... protocol="abfs", ... account_name="mydatalake", ... tenant_id="tenant123", ... client_id="client123", ... client_secret="secret123" ... )

Simple connection string auth

options = AzureStorageOptions( ... protocol="az", ... connection_string="DefaultEndpoints..." ... )

Functions

fsspeckit.storage_options.AzureStorageOptions.from_env classmethod

from_env() -> AzureStorageOptions

Create storage options from environment variables.

Reads standard Azure environment variables: - AZURE_STORAGE_ACCOUNT_NAME - AZURE_STORAGE_ACCOUNT_KEY - AZURE_STORAGE_CONNECTION_STRING - AZURE_TENANT_ID - AZURE_CLIENT_ID - AZURE_CLIENT_SECRET - AZURE_STORAGE_SAS_TOKEN

Returns:

Name	Type	Description
AzureStorageOptions	AzureStorageOptions	Configured storage options

Example

With environment variables set:

options = AzureStorageOptions.from_env() print(options.account_name) # From AZURE_STORAGE_ACCOUNT_NAME 'mystorageacct'

Source code in src/fsspeckit/storage_options/cloud.py

@classmethod
def from_env(cls) -> "AzureStorageOptions":
    """Create storage options from environment variables.

    Reads standard Azure environment variables:
    - AZURE_STORAGE_ACCOUNT_NAME
    - AZURE_STORAGE_ACCOUNT_KEY
    - AZURE_STORAGE_CONNECTION_STRING
    - AZURE_TENANT_ID
    - AZURE_CLIENT_ID
    - AZURE_CLIENT_SECRET
    - AZURE_STORAGE_SAS_TOKEN

    Returns:
        AzureStorageOptions: Configured storage options

    Example:
        >>> # With environment variables set:
        >>> options = AzureStorageOptions.from_env()
        >>> print(options.account_name)  # From AZURE_STORAGE_ACCOUNT_NAME
        'mystorageacct'
    """
    return cls(
        protocol=os.getenv("AZURE_STORAGE_PROTOCOL", "az"),
        account_name=os.getenv("AZURE_STORAGE_ACCOUNT_NAME"),
        account_key=os.getenv("AZURE_STORAGE_ACCOUNT_KEY"),
        connection_string=os.getenv("AZURE_STORAGE_CONNECTION_STRING"),
        tenant_id=os.getenv("AZURE_TENANT_ID"),
        client_id=os.getenv("AZURE_CLIENT_ID"),
        client_secret=os.getenv("AZURE_CLIENT_SECRET"),
        sas_token=os.getenv("AZURE_STORAGE_SAS_TOKEN"),
    )

fsspeckit.storage_options.AzureStorageOptions.to_env

to_env() -> None

Export options to environment variables.

Sets standard Azure environment variables.

Example

options = AzureStorageOptions( ... protocol="az", ... account_name="mystorageacct", ... account_key="key123" ... ) options.to_env() print(os.getenv("AZURE_STORAGE_ACCOUNT_NAME")) 'mystorageacct'

Source code in src/fsspeckit/storage_options/cloud.py

def to_env(self) -> None:
    """Export options to environment variables.

    Sets standard Azure environment variables.

    Example:
        >>> options = AzureStorageOptions(
        ...     protocol="az",
        ...     account_name="mystorageacct",
        ...     account_key="key123"
        ... )
        >>> options.to_env()
        >>> print(os.getenv("AZURE_STORAGE_ACCOUNT_NAME"))
        'mystorageacct'
    """
    env = {
        "AZURE_STORAGE_PROTOCOL": self._normalized_protocol,
        "AZURE_STORAGE_ACCOUNT_NAME": self.account_name,
        "AZURE_STORAGE_ACCOUNT_KEY": self.account_key,
        "AZURE_STORAGE_CONNECTION_STRING": self.connection_string,
        "AZURE_TENANT_ID": self.tenant_id,
        "AZURE_CLIENT_ID": self.client_id,
        "AZURE_CLIENT_SECRET": self.client_secret,
        "AZURE_STORAGE_SAS_TOKEN": self.sas_token,
    }
    env = {k: str(v) for k, v in env.items() if v is not None}
    os.environ.update(env)  # type: ignore[arg-type]

fsspeckit.storage_options.AzureStorageOptions.to_fsspec_kwargs

to_fsspec_kwargs() -> dict

Convert options to Azure-compatible fsspec kwargs.

Source code in src/fsspeckit/storage_options/cloud.py

def to_fsspec_kwargs(self) -> dict:
    """Convert options to Azure-compatible fsspec kwargs."""

    kwargs = {
        "account_name": self.account_name,
        "account_key": self.account_key,
        "connection_string": self.connection_string,
        "tenant_id": self.tenant_id,
        "client_id": self.client_id,
        "client_secret": self.client_secret,
        "sas_token": self.sas_token,
    }
    return {k: v for k, v in kwargs.items() if v is not None}

fsspeckit.storage_options.GitHubStorageOptions

Bases: BaseStorageOptions

GitHub repository storage configuration options.

Provides access to files in GitHub repositories with support for: - Public and private repositories - Branch/tag/commit selection - Token-based authentication - Custom GitHub Enterprise instances

Attributes:

Name	Type	Description
protocol	str	Always "github" for GitHub storage
org	str	Organization or user name
repo	str	Repository name
ref	str	Git reference (branch, tag, or commit SHA)
token	str	GitHub personal access token
api_url	str	Custom GitHub API URL for enterprise instances

Example

# Public repository
options = GitHubStorageOptions(
    org="microsoft",
    repo="vscode",
    ref="main",
)

# Private repository
options = GitHubStorageOptions(
    org="myorg",
    repo="private-repo",
    token="ghp_xxxx",
    ref="develop",
)

# Enterprise instance
options = GitHubStorageOptions(
    org="company",
    repo="internal",
    api_url="https://github.company.com/api/v3",
    token="ghp_xxxx",
)

Functions

fsspeckit.storage_options.GitHubStorageOptions.from_env classmethod

from_env() -> GitHubStorageOptions

Create storage options from environment variables.

Reads standard GitHub environment variables: - GITHUB_ORG: Organization or user name - GITHUB_REPO: Repository name - GITHUB_REF: Git reference - GITHUB_TOKEN: Personal access token - GITHUB_API_URL: Custom API URL

Returns:

Name	Type	Description
GitHubStorageOptions	GitHubStorageOptions	Configured storage options

Example

# With environment variables set:
options = GitHubStorageOptions.from_env()
print(options.org)  # From GITHUB_ORG
# 'microsoft'

Source code in src/fsspeckit/storage_options/git.py

@classmethod
def from_env(cls) -> "GitHubStorageOptions":
    """Create storage options from environment variables.

    Reads standard GitHub environment variables:
    - GITHUB_ORG: Organization or user name
    - GITHUB_REPO: Repository name
    - GITHUB_REF: Git reference
    - GITHUB_TOKEN: Personal access token
    - GITHUB_API_URL: Custom API URL

    Returns:
        GitHubStorageOptions: Configured storage options

    Example:
        ```python
        # With environment variables set:
        options = GitHubStorageOptions.from_env()
        print(options.org)  # From GITHUB_ORG
        # 'microsoft'
        ```
    """
    return cls(
        protocol="github",
        org=os.getenv("GITHUB_ORG"),
        repo=os.getenv("GITHUB_REPO"),
        ref=os.getenv("GITHUB_REF"),
        token=os.getenv("GITHUB_TOKEN"),
        api_url=os.getenv("GITHUB_API_URL"),
    )

fsspeckit.storage_options.GitHubStorageOptions.to_env

to_env() -> None

Export options to environment variables.

Sets standard GitHub environment variables.

Example

options = GitHubStorageOptions(
    org="microsoft",
    repo="vscode",
    token="ghp_xxxx",
)
options.to_env()
print(os.getenv("GITHUB_ORG"))
# 'microsoft'

Source code in src/fsspeckit/storage_options/git.py

def to_env(self) -> None:
    """Export options to environment variables.

    Sets standard GitHub environment variables.

    Example:
        ```python
        options = GitHubStorageOptions(
            org="microsoft",
            repo="vscode",
            token="ghp_xxxx",
        )
        options.to_env()
        print(os.getenv("GITHUB_ORG"))
        # 'microsoft'
        ```
    """
    env = {
        "GITHUB_ORG": self.org,
        "GITHUB_REPO": self.repo,
        "GITHUB_REF": self.ref,
        "GITHUB_TOKEN": self.token,
        "GITHUB_API_URL": self.api_url,
    }
    env = {k: v for k, v in env.items() if v is not None}
    os.environ.update(env)

fsspeckit.storage_options.GitHubStorageOptions.to_fsspec_kwargs

to_fsspec_kwargs() -> dict

Convert options to fsspec filesystem arguments.

Returns:

Name	Type	Description
dict	dict	Arguments suitable for GitHubFileSystem

Example

options = GitHubStorageOptions( ... org="microsoft", ... repo="vscode", ... token="ghp_xxxx" ... ) kwargs = options.to_fsspec_kwargs() fs = filesystem("github", **kwargs)

Source code in src/fsspeckit/storage_options/git.py

def to_fsspec_kwargs(self) -> dict:
    """Convert options to fsspec filesystem arguments.

    Returns:
        dict: Arguments suitable for GitHubFileSystem

    Example:
        >>> options = GitHubStorageOptions(
        ...     org="microsoft",
        ...     repo="vscode",
        ...     token="ghp_xxxx"
        ... )
        >>> kwargs = options.to_fsspec_kwargs()
        >>> fs = filesystem("github", **kwargs)
    """
    kwargs = {
        "org": self.org,
        "repo": self.repo,
        "ref": self.ref,
        "token": self.token,
        "api_url": self.api_url,
    }
    return {k: v for k, v in kwargs.items() if v is not None}

fsspeckit.storage_options.GitLabStorageOptions

Bases: BaseStorageOptions

GitLab repository storage configuration options.

Provides access to files in GitLab repositories with support for: - Public and private repositories - Self-hosted GitLab instances - Project ID or name-based access - Branch/tag/commit selection - Token-based authentication

Attributes:

Name	Type	Description
protocol	str	Always "gitlab" for GitLab storage
base_url	str	GitLab instance URL, defaults to gitlab.com
project_id	Union[str, int]	Project ID number
project_name	str	Project name/path
ref	str	Git reference (branch, tag, or commit SHA)
token	str	GitLab personal access token
api_version	str	API version to use
timeout	float	Request timeout in seconds (must be positive, max 3600)
max_pages	int	Maximum number of pages to fetch (default 1000, max 10000)

Example

# Public project on gitlab.com
options = GitLabStorageOptions(
    project_name="group/project",
    ref="main",
)

# Private project with token
options = GitLabStorageOptions(
    project_id=12345,
    token="glpat_xxxx",
    ref="develop",
)

# Self-hosted instance
options = GitLabStorageOptions(
    base_url="https://gitlab.company.com",
    project_name="internal/project",
    token="glpat_xxxx",
)

Functions

fsspeckit.storage_options.GitLabStorageOptions.__post_init__

__post_init__() -> None

Validate GitLab configuration after initialization.

Ensures either project_id or project_name is provided and validates timeout and max_pages parameters.

Raises:

Type	Description
ValueError	If neither project_id nor project_name is provided, or if timeout/max_pages values are invalid

Source code in src/fsspeckit/storage_options/git.py

def __post_init__(self) -> None:
    """Validate GitLab configuration after initialization.

    Ensures either project_id or project_name is provided and validates
    timeout and max_pages parameters.

    Raises:
        ValueError: If neither project_id nor project_name is provided,
                   or if timeout/max_pages values are invalid
    """
    if self.project_id is None and self.project_name is None:
        raise ValueError("Either project_id or project_name must be provided")

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

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

fsspeckit.storage_options.GitLabStorageOptions.from_env classmethod

from_env() -> GitLabStorageOptions

Create storage options from environment variables.

Reads standard GitLab environment variables: - GITLAB_URL: Instance URL - GITLAB_PROJECT_ID: Project ID - GITLAB_PROJECT_NAME: Project name/path - GITLAB_REF: Git reference - GITLAB_TOKEN: Personal access token - GITLAB_API_VERSION: API version - GITLAB_TIMEOUT: Request timeout in seconds - GITLAB_MAX_PAGES: Maximum number of pages to fetch

Returns:

Name	Type	Description
GitLabStorageOptions	GitLabStorageOptions	Configured storage options

Example

# With environment variables set:
options = GitLabStorageOptions.from_env()
print(options.project_id)  # From GITLAB_PROJECT_ID
# '12345'

Source code in src/fsspeckit/storage_options/git.py

@classmethod
def from_env(cls) -> "GitLabStorageOptions":
    """Create storage options from environment variables.

    Reads standard GitLab environment variables:
    - GITLAB_URL: Instance URL
    - GITLAB_PROJECT_ID: Project ID
    - GITLAB_PROJECT_NAME: Project name/path
    - GITLAB_REF: Git reference
    - GITLAB_TOKEN: Personal access token
    - GITLAB_API_VERSION: API version
    - GITLAB_TIMEOUT: Request timeout in seconds
    - GITLAB_MAX_PAGES: Maximum number of pages to fetch

    Returns:
        GitLabStorageOptions: Configured storage options

    Example:
        ```python
        # With environment variables set:
        options = GitLabStorageOptions.from_env()
        print(options.project_id)  # From GITLAB_PROJECT_ID
        # '12345'
        ```
    """
    return cls(
        protocol="gitlab",
        base_url=os.getenv("GITLAB_URL", "https://gitlab.com"),
        project_id=os.getenv("GITLAB_PROJECT_ID"),
        project_name=os.getenv("GITLAB_PROJECT_NAME"),
        ref=os.getenv("GITLAB_REF"),
        token=os.getenv("GITLAB_TOKEN"),
        api_version=os.getenv("GITLAB_API_VERSION", "v4"),
        timeout=float(os.getenv("GITLAB_TIMEOUT")) if os.getenv("GITLAB_TIMEOUT") else None,
        max_pages=int(os.getenv("GITLAB_MAX_PAGES")) if os.getenv("GITLAB_MAX_PAGES") else None,
    )

fsspeckit.storage_options.GitLabStorageOptions.to_env

to_env() -> None

Export options to environment variables.

Sets standard GitLab environment variables.

Example

options = GitLabStorageOptions(
    project_id=12345,
    token="glpat_xxxx",
)
options.to_env()
print(os.getenv("GITLAB_PROJECT_ID"))
# '12345'

Source code in src/fsspeckit/storage_options/git.py

def to_env(self) -> None:
    """Export options to environment variables.

    Sets standard GitLab environment variables.

    Example:
        ```python
        options = GitLabStorageOptions(
            project_id=12345,
            token="glpat_xxxx",
        )
        options.to_env()
        print(os.getenv("GITLAB_PROJECT_ID"))
        # '12345'
        ```
    """
    env = {
        "GITLAB_URL": self.base_url,
        "GITLAB_PROJECT_ID": str(self.project_id) if self.project_id else None,
        "GITLAB_PROJECT_NAME": self.project_name,
        "GITLAB_REF": self.ref,
        "GITLAB_TOKEN": self.token,
        "GITLAB_API_VERSION": self.api_version,
        "GITLAB_TIMEOUT": str(self.timeout) if self.timeout is not None else None,
        "GITLAB_MAX_PAGES": str(self.max_pages) if self.max_pages is not None else None,
    }
    env = {k: v for k, v in env.items() if v is not None}
    os.environ.update(env)

fsspeckit.storage_options.GitLabStorageOptions.to_fsspec_kwargs

to_fsspec_kwargs() -> dict

Convert options to fsspec filesystem arguments.

Returns:

Name	Type	Description
dict	dict	Arguments suitable for GitLabFileSystem

Example

options = GitLabStorageOptions(
    project_id=12345,
    token="glpat_xxxx",
)
kwargs = options.to_fsspec_kwargs()
fs = filesystem("gitlab", **kwargs)

Source code in src/fsspeckit/storage_options/git.py

def to_fsspec_kwargs(self) -> dict:
    """Convert options to fsspec filesystem arguments.

    Returns:
        dict: Arguments suitable for GitLabFileSystem

    Example:
        ```python
        options = GitLabStorageOptions(
            project_id=12345,
            token="glpat_xxxx",
        )
        kwargs = options.to_fsspec_kwargs()
        fs = filesystem("gitlab", **kwargs)
        ```
    """
    kwargs = {
        "base_url": self.base_url,
        "project_id": self.project_id,
        "project_name": self.project_name,
        "ref": self.ref,
        "token": self.token,
        "api_version": self.api_version,
        "timeout": self.timeout,
        "max_pages": self.max_pages,
    }
    return {k: v for k, v in kwargs.items() if v is not None}

Functions

fsspeckit.storage_options.from_dict

from_dict(
    protocol: str, storage_options: dict
) -> BaseStorageOptions

Create appropriate storage options instance from dictionary.

Factory function that creates the correct storage options class based on protocol.

Parameters:

Name	Type	Description	Default
protocol	str	Storage protocol identifier (e.g., "s3", "gs", "file")	required
storage_options	dict	Dictionary of configuration options	required

Returns:

Name	Type	Description
BaseStorageOptions	BaseStorageOptions	Appropriate storage options instance

Raises:

Type	Description
ValueError	If protocol is not supported

Example

# Create S3 options
options = from_dict(
    "s3",
    {
        "access_key_id": "KEY",
        "secret_access_key": "SECRET",
    },
)
print(type(options).__name__)
# 'AwsStorageOptions'

Source code in src/fsspeckit/storage_options/core.py

def from_dict(protocol: str, storage_options: dict) -> BaseStorageOptions:
    """Create appropriate storage options instance from dictionary.

    Factory function that creates the correct storage options class based on protocol.

    Args:
        protocol: Storage protocol identifier (e.g., "s3", "gs", "file")
        storage_options: Dictionary of configuration options

    Returns:
        BaseStorageOptions: Appropriate storage options instance

    Raises:
        ValueError: If protocol is not supported

    Example:
        ```python
        # Create S3 options
        options = from_dict(
            "s3",
            {
                "access_key_id": "KEY",
                "secret_access_key": "SECRET",
            },
        )
        print(type(options).__name__)
        # 'AwsStorageOptions'
        ```
    """
    if protocol == "s3":
        # Handle anon -> anonymous parameter translation
        if "anon" in storage_options:
            storage_options = storage_options.copy()
            storage_options["anonymous"] = storage_options.pop("anon")

        if (
            "profile" in storage_options
            or "key" in storage_options
            or "secret" in storage_options
        ):
            return AwsStorageOptions.create(**storage_options)
        return AwsStorageOptions(**storage_options)
    elif protocol in ["az", "abfs", "adl"]:
        return AzureStorageOptions(**storage_options)
    elif protocol in ["gs", "gcs"]:
        return GcsStorageOptions(**storage_options)
    elif protocol == "github":
        return GitHubStorageOptions(**storage_options)
    elif protocol == "gitlab":
        return GitLabStorageOptions(**storage_options)
    elif protocol == "file":
        return LocalStorageOptions(**storage_options)
    else:
        raise ValueError(f"Unsupported protocol: {protocol}")

fsspeckit.storage_options.from_env

from_env(protocol: str) -> BaseStorageOptions

Create storage options from environment variables.

Factory function that creates and configures storage options from protocol-specific environment variables.

Parameters:

Name	Type	Description	Default
protocol	str	Storage protocol identifier (e.g., "s3", "github")	required

Returns:

Name	Type	Description
BaseStorageOptions	BaseStorageOptions	Configured storage options instance

Raises:

Type	Description
ValueError	If protocol is not supported

Example

# With AWS credentials in environment
options = from_env("s3")
print(options.access_key_id)  # From AWS_ACCESS_KEY_ID
# 'AKIAXXXXXX'

Source code in src/fsspeckit/storage_options/core.py

def from_env(protocol: str) -> BaseStorageOptions:
    """Create storage options from environment variables.

    Factory function that creates and configures storage options from
    protocol-specific environment variables.

    Args:
        protocol: Storage protocol identifier (e.g., "s3", "github")

    Returns:
        BaseStorageOptions: Configured storage options instance

    Raises:
        ValueError: If protocol is not supported

    Example:
        ```python
        # With AWS credentials in environment
        options = from_env("s3")
        print(options.access_key_id)  # From AWS_ACCESS_KEY_ID
        # 'AKIAXXXXXX'
        ```
    """
    if protocol == "s3":
        return AwsStorageOptions.from_env()
    elif protocol in ["az", "abfs", "adl"]:
        return AzureStorageOptions.from_env()
    elif protocol in ["gs", "gcs"]:
        return GcsStorageOptions.from_env()
    elif protocol == "github":
        return GitHubStorageOptions.from_env()
    elif protocol == "gitlab":
        return GitLabStorageOptions.from_env()
    elif protocol == "file":
        return LocalStorageOptions()
    else:
        raise ValueError(f"Unsupported protocol: {protocol}")

fsspeckit.storage_options.merge_storage_options

merge_storage_options(
    *options: BaseStorageOptions | dict | None,
    overwrite: bool = True,
) -> BaseStorageOptions

Merge multiple storage options into a single configuration.

Combines options from multiple sources with control over precedence.

Parameters:

Name	Type	Description	Default
*options	BaseStorageOptions | dict | None	Storage options to merge. Can be: - BaseStorageOptions instances - Dictionaries of options - None values (ignored)	()
overwrite	bool	Whether later options override earlier ones	True

Returns:

Name	Type	Description
BaseStorageOptions	BaseStorageOptions	Combined storage options

Example

# Merge with overwrite
base = AwsStorageOptions(
    region="us-east-1",
    access_key_id="OLD_KEY",
)
override = {"access_key_id": "NEW_KEY"}
merged = merge_storage_options(base, override)
print(merged.access_key_id)
# 'NEW_KEY'

# Preserve existing values
merged = merge_storage_options(
    base,
    override,
    overwrite=False,
)
print(merged.access_key_id)
# 'OLD_KEY'

Source code in src/fsspeckit/storage_options/core.py

def merge_storage_options(
    *options: BaseStorageOptions | dict | None, overwrite: bool = True
) -> BaseStorageOptions:
    """Merge multiple storage options into a single configuration.

    Combines options from multiple sources with control over precedence.

    Args:
        *options: Storage options to merge. Can be:
            - BaseStorageOptions instances
            - Dictionaries of options
            - None values (ignored)
        overwrite: Whether later options override earlier ones

    Returns:
        BaseStorageOptions: Combined storage options

    Example:
        ```python
        # Merge with overwrite
        base = AwsStorageOptions(
            region="us-east-1",
            access_key_id="OLD_KEY",
        )
        override = {"access_key_id": "NEW_KEY"}
        merged = merge_storage_options(base, override)
        print(merged.access_key_id)
        # 'NEW_KEY'

        # Preserve existing values
        merged = merge_storage_options(
            base,
            override,
            overwrite=False,
        )
        print(merged.access_key_id)
        # 'OLD_KEY'
        ```
    """
    result = {}
    protocol = None

    for opts in options:
        if opts is None:
            continue
        if isinstance(opts, BaseStorageOptions):
            opts = opts.to_dict(with_protocol=True)
        if not protocol and "protocol" in opts:
            protocol = opts["protocol"]
        for k, v in opts.items():
            if overwrite or k not in result:
                result[k] = v

    if not protocol:
        protocol = "file"
    return from_dict(protocol, result)

fsspeckit.storage_options.infer_protocol_from_uri

infer_protocol_from_uri(uri: str) -> str

Infer the storage protocol from a URI string.

Analyzes the URI to determine the appropriate storage protocol based on the scheme or path format.

Parameters:

Name	Type	Description	Default
uri	str	URI or path string to analyze. Examples: - "s3://bucket/path" - "gs://bucket/path" - "github://org/repo" - "/local/path"	required

Returns:

Name	Type	Description
str	str	Inferred protocol identifier

Example

# S3 protocol
infer_protocol_from_uri("s3://my-bucket/data")
# 's3'

# Local file
infer_protocol_from_uri("/home/user/data")
# 'file'

# GitHub repository
infer_protocol_from_uri("github://microsoft/vscode")
# 'github'

Source code in src/fsspeckit/storage_options/core.py

def infer_protocol_from_uri(uri: str) -> str:
    """Infer the storage protocol from a URI string.

    Analyzes the URI to determine the appropriate storage protocol based on
    the scheme or path format.

    Args:
        uri: URI or path string to analyze. Examples:
            - "s3://bucket/path"
            - "gs://bucket/path"
            - "github://org/repo"
            - "/local/path"

    Returns:
        str: Inferred protocol identifier

    Example:
        ```python
        # S3 protocol
        infer_protocol_from_uri("s3://my-bucket/data")
        # 's3'

        # Local file
        infer_protocol_from_uri("/home/user/data")
        # 'file'

        # GitHub repository
        infer_protocol_from_uri("github://microsoft/vscode")
        # 'github'
        ```
    """
    if uri.startswith("s3://"):
        return "s3"
    elif uri.startswith("gs://") or uri.startswith("gcs://"):
        return "gs"
    elif uri.startswith("github://"):
        return "github"
    elif uri.startswith("gitlab://"):
        return "gitlab"
    elif uri.startswith(("az://", "abfs://", "adl://")):
        return uri.split("://")[0]
    else:
        return "file"

fsspeckit.storage_options.storage_options_from_uri

storage_options_from_uri(uri: str) -> BaseStorageOptions

Create storage options instance from a URI string.

Infers the protocol and extracts relevant configuration from the URI to create appropriate storage options.

Parameters:

Name	Type	Description	Default
uri	str	URI string containing protocol and optional configuration. Examples: - "s3://bucket/path" - "gs://project/bucket/path" - "github://org/repo"	required

Returns:

Name	Type	Description
BaseStorageOptions	BaseStorageOptions	Configured storage options instance

Example

# S3 options
opts = storage_options_from_uri("s3://my-bucket/data")
print(opts.protocol)
# 's3'

# GitHub options
opts = storage_options_from_uri("github://microsoft/vscode")
print(opts.org)
# 'microsoft'
print(opts.repo)
# 'vscode'

Source code in src/fsspeckit/storage_options/core.py

def storage_options_from_uri(uri: str) -> BaseStorageOptions:
    """Create storage options instance from a URI string.

    Infers the protocol and extracts relevant configuration from the URI
    to create appropriate storage options.

    Args:
        uri: URI string containing protocol and optional configuration.
            Examples:
            - "s3://bucket/path"
            - "gs://project/bucket/path"
            - "github://org/repo"

    Returns:
        BaseStorageOptions: Configured storage options instance

    Example:
        ```python
        # S3 options
        opts = storage_options_from_uri("s3://my-bucket/data")
        print(opts.protocol)
        # 's3'

        # GitHub options
        opts = storage_options_from_uri("github://microsoft/vscode")
        print(opts.org)
        # 'microsoft'
        print(opts.repo)
        # 'vscode'
        ```
    """
    protocol = infer_protocol_from_uri(uri)
    options = infer_storage_options(uri)

    if protocol == "s3":
        return AwsStorageOptions(protocol=protocol, **options)
    elif protocol in ["gs", "gcs"]:
        return GcsStorageOptions(protocol=protocol, **options)
    elif protocol == "github":
        parts = uri.replace("github://", "").split("/")
        return GitHubStorageOptions(
            protocol=protocol, org=parts[0], repo=parts[1] if len(parts) > 1 else None
        )
    elif protocol == "gitlab":
        parts = uri.replace("gitlab://", "").split("/")
        return GitLabStorageOptions(
            protocol=protocol, project_name=parts[-1] if parts else None
        )
    elif protocol in ["az", "abfs", "adl"]:
        return AzureStorageOptions(protocol=protocol, **options)
    else:
        return LocalStorageOptions()