Skip to content

Storage API Reference

cs_copilot.storage

Cs_copilot Storage Module

Provides unified storage abstraction for local files and S3/MinIO. Handles session management and file operations transparently.

Main Components:

  • S3: Storage client class for file operations
  • SESSION_ID: Current session identifier
  • S3Config: Configuration dataclass for S3 settings

Usage:

from cs_copilot.storage import S3

# Read a file
with S3.open("data/compounds.csv", "r") as f:
    df = pd.read_csv(f)

# Write a file
with S3.open("results/output.csv", "w") as f:
    df.to_csv(f)

# Get S3 path
path = S3.path("results/model.pkl.gz")

S3

Unified storage client for S3/MinIO and local file operations.

All file operations are scoped to the current session by default, unless an absolute S3 URL or local path is provided.

Class Attributes:

prefix : str Session-scoped prefix for all relative paths

Methods:

path(rel: str) -> str Convert a relative path to an S3 URL

open(rel: str, mode: str = "rb") Open a file for reading or writing

Examples:

Write to session-scoped S3 path

with S3.open("results.csv", "w") as f: ... f.write("data")

Read from absolute S3 URL

with S3.open("s3://bucket/key.csv", "r") as f: ... data = f.read()

Read from local file

with S3.open("/tmp/local.csv", "r") as f: ... data = f.read()

Source code in src/cs_copilot/storage/client.py 
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
class S3:
    """
    Unified storage client for S3/MinIO and local file operations.

    All file operations are scoped to the current session by default,
    unless an absolute S3 URL or local path is provided.

    Class Attributes:
    -----------------
    prefix : str
        Session-scoped prefix for all relative paths

    Methods:
    --------
    path(rel: str) -> str
        Convert a relative path to an S3 URL

    open(rel: str, mode: str = "rb")
        Open a file for reading or writing

    Examples:
    ---------
    >>> # Write to session-scoped S3 path
    >>> with S3.open("results.csv", "w") as f:
    ...     f.write("data")

    >>> # Read from absolute S3 URL
    >>> with S3.open("s3://bucket/key.csv", "r") as f:
    ...     data = f.read()

    >>> # Read from local file
    >>> with S3.open("/tmp/local.csv", "r") as f:
    ...     data = f.read()
    """

    prefix = f"sessions/{SESSION_ID}"
    logger.info(f"Initialized S3 client with SESSION_ID: {SESSION_ID}")

    @classmethod
    def path(cls, rel: str) -> str:
        """
        Convert a relative path to an S3 URL.

        If an absolute S3 URL is provided (starts with s3://),
        it is returned unchanged.

        Args:
            rel: Relative path or absolute S3 URL

        Returns:
            str: Full S3 URL

        Examples:
        ---------
        >>> S3.path("data.csv")
        's3://chatbot-assets/sessions/20250121-123456-abc123/data.csv'

        >>> S3.path("s3://mybucket/data.csv")
        's3://mybucket/data.csv'
        """
        # If an absolute S3 URL is provided, pass it through unchanged
        if isinstance(rel, str) and rel.startswith("s3://"):
            return rel

        # Always produce a URL; works with fsspec & pandas
        config = get_s3_config()
        return f"s3://{config.bucket_name}/{cls.prefix}/{rel}".strip("/")

    @classmethod
    def open(cls, rel: str, mode: str = "rb"):
        """
        Open a file for reading or writing.

        Supports three types of paths:
        1. Absolute S3 URLs (s3://...)
        2. Local absolute paths (/ or file://)
        3. Relative paths (scoped to current session)

        Args:
            rel: File path (relative, absolute, or S3 URL)
            mode: File mode ('r', 'w', 'rb', 'wb', etc.)

        Returns:
            File-like object opened with the specified mode

        Examples:
        ---------
        >>> # Read from session-scoped S3 path
        >>> with S3.open("data.csv", "r") as f:
        ...     df = pd.read_csv(f)

        >>> # Write to session-scoped S3 path
        >>> with S3.open("output.csv", "w") as f:
        ...     df.to_csv(f)

        >>> # Read from absolute S3 URL
        >>> with S3.open("s3://bucket/data.csv", "r") as f:
        ...     df = pd.read_csv(f)

        >>> # Read from local file
        >>> with S3.open("/tmp/data.csv", "r") as f:
        ...     df = pd.read_csv(f)
        """
        config = get_s3_config()
        use_s3 = is_s3_enabled()

        # 1) Absolute S3 URL → open as-is
        if isinstance(rel, str) and rel.startswith("s3://"):
            return fsspec.open(rel, mode=mode, **config.to_storage_options())

        # 2) Explicit local files (absolute paths or file://) → always open locally
        if isinstance(rel, str) and (rel.startswith("/") or rel.startswith("file://")):
            if rel.startswith("file://"):
                return fsspec.open(rel, mode=mode)
            return builtins.open(rel, mode)

        # 3) If S3 is disabled, allow relative local file access
        if not use_s3:
            if isinstance(rel, str) and rel.startswith("file://"):
                return fsspec.open(rel, mode=mode)
            return builtins.open(rel, mode)

        # 4) Otherwise treat as a key relative to the session prefix and force S3
        return fsspec.open(cls.path(rel), mode=mode, **config.to_storage_options())

path(rel) classmethod

Convert a relative path to an S3 URL.

If an absolute S3 URL is provided (starts with s3://), it is returned unchanged.

Parameters:

Name Type Description Default
rel str

Relative path or absolute S3 URL

 required

Returns:

Name Type Description
str str

Full S3 URL
Examples:

S3.path("data.csv") 's3://chatbot-assets/sessions/20250121-123456-abc123/data.csv'

S3.path("s3://mybucket/data.csv") 's3://mybucket/data.csv'

Source code in src/cs_copilot/storage/client.py 
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
@classmethod
def path(cls, rel: str) -> str:
    """
    Convert a relative path to an S3 URL.

    If an absolute S3 URL is provided (starts with s3://),
    it is returned unchanged.

    Args:
        rel: Relative path or absolute S3 URL

    Returns:
        str: Full S3 URL

    Examples:
    ---------
    >>> S3.path("data.csv")
    's3://chatbot-assets/sessions/20250121-123456-abc123/data.csv'

    >>> S3.path("s3://mybucket/data.csv")
    's3://mybucket/data.csv'
    """
    # If an absolute S3 URL is provided, pass it through unchanged
    if isinstance(rel, str) and rel.startswith("s3://"):
        return rel

    # Always produce a URL; works with fsspec & pandas
    config = get_s3_config()
    return f"s3://{config.bucket_name}/{cls.prefix}/{rel}".strip("/")

open(rel, mode='rb') classmethod

Open a file for reading or writing.

Supports three types of paths: 1. Absolute S3 URLs (s3://...) 2. Local absolute paths (/ or file://) 3. Relative paths (scoped to current session)

Parameters:

Name Type Description Default
rel str

File path (relative, absolute, or S3 URL)

 required
mode str

File mode ('r', 'w', 'rb', 'wb', etc.)

 'rb'

Returns:

Type Description

File-like object opened with the specified mode
Examples:
Read from session-scoped S3 path

with S3.open("data.csv", "r") as f: ... df = pd.read_csv(f)

Write to session-scoped S3 path

with S3.open("output.csv", "w") as f: ... df.to_csv(f)

Read from absolute S3 URL

with S3.open("s3://bucket/data.csv", "r") as f: ... df = pd.read_csv(f)

Read from local file

with S3.open("/tmp/data.csv", "r") as f: ... df = pd.read_csv(f)

Source code in src/cs_copilot/storage/client.py 
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
@classmethod
def open(cls, rel: str, mode: str = "rb"):
    """
    Open a file for reading or writing.

    Supports three types of paths:
    1. Absolute S3 URLs (s3://...)
    2. Local absolute paths (/ or file://)
    3. Relative paths (scoped to current session)

    Args:
        rel: File path (relative, absolute, or S3 URL)
        mode: File mode ('r', 'w', 'rb', 'wb', etc.)

    Returns:
        File-like object opened with the specified mode

    Examples:
    ---------
    >>> # Read from session-scoped S3 path
    >>> with S3.open("data.csv", "r") as f:
    ...     df = pd.read_csv(f)

    >>> # Write to session-scoped S3 path
    >>> with S3.open("output.csv", "w") as f:
    ...     df.to_csv(f)

    >>> # Read from absolute S3 URL
    >>> with S3.open("s3://bucket/data.csv", "r") as f:
    ...     df = pd.read_csv(f)

    >>> # Read from local file
    >>> with S3.open("/tmp/data.csv", "r") as f:
    ...     df = pd.read_csv(f)
    """
    config = get_s3_config()
    use_s3 = is_s3_enabled()

    # 1) Absolute S3 URL → open as-is
    if isinstance(rel, str) and rel.startswith("s3://"):
        return fsspec.open(rel, mode=mode, **config.to_storage_options())

    # 2) Explicit local files (absolute paths or file://) → always open locally
    if isinstance(rel, str) and (rel.startswith("/") or rel.startswith("file://")):
        if rel.startswith("file://"):
            return fsspec.open(rel, mode=mode)
        return builtins.open(rel, mode)

    # 3) If S3 is disabled, allow relative local file access
    if not use_s3:
        if isinstance(rel, str) and rel.startswith("file://"):
            return fsspec.open(rel, mode=mode)
        return builtins.open(rel, mode)

    # 4) Otherwise treat as a key relative to the session prefix and force S3
    return fsspec.open(cls.path(rel), mode=mode, **config.to_storage_options())

S3Config dataclass

Configuration for S3/MinIO storage.

Source code in src/cs_copilot/storage/config.py 
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
@dataclass
class S3Config:
    """Configuration for S3/MinIO storage."""

    endpoint_url: Optional[str]
    access_key_id: str
    secret_access_key: str
    bucket_name: str
    region_name: str = "us-east-1"
    use_s3: bool = True

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

        Environment Variables:
        ----------------------
        Endpoint (one of):
            - MINIO_ENDPOINT
            - MINIO_ENDPOINT_URL
            - S3_ENDPOINT_URL

        Access Key (one of):
            - MINIO_ACCESS_KEY
            - AWS_ACCESS_KEY_ID

        Secret Key (one of):
            - MINIO_SECRET_KEY
            - AWS_SECRET_ACCESS_KEY

        Bucket Name (one of):
            - ASSETS_BUCKET
            - S3_BUCKET_NAME

        Other:
            - AWS_REGION (default: us-east-1)
            - USE_S3 (default: true)

        Returns:
            S3Config: Configuration instance
        """
        # Endpoint fallbacks: MINIO_ENDPOINT, MINIO_ENDPOINT_URL, S3_ENDPOINT_URL
        endpoint = (
            os.getenv("MINIO_ENDPOINT")
            or os.getenv("MINIO_ENDPOINT_URL")
            or os.getenv("S3_ENDPOINT_URL")
            or None  # When unset, let s3fs/botocore pick AWS endpoint
        )

        # Access key fallbacks: MINIO_ACCESS_KEY, AWS_ACCESS_KEY_ID
        access_key = os.getenv("MINIO_ACCESS_KEY") or os.getenv("AWS_ACCESS_KEY_ID") or "minioadmin"

        # Secret key fallbacks: MINIO_SECRET_KEY, AWS_SECRET_ACCESS_KEY
        secret_key = (
            os.getenv("MINIO_SECRET_KEY") or os.getenv("AWS_SECRET_ACCESS_KEY") or "minioadmin"
        )

        # Bucket name fallbacks: ASSETS_BUCKET, S3_BUCKET_NAME
        bucket_name = os.getenv("ASSETS_BUCKET") or os.getenv("S3_BUCKET_NAME") or "chatbot-assets"

        # Region
        region_name = os.getenv("AWS_REGION", "us-east-1")

        # Use S3 flag
        use_s3 = os.getenv("USE_S3", "true").lower() == "true"

        return cls(
            endpoint_url=endpoint,
            access_key_id=access_key,
            secret_access_key=secret_key,
            bucket_name=bucket_name,
            region_name=region_name,
            use_s3=use_s3,
        )

    def to_storage_options(self) -> dict:
        """
        Convert config to fsspec/s3fs storage options.

        Returns:
            dict: Storage options for fsspec.open()
        """
        opts = {
            "key": self.access_key_id,
            "secret": self.secret_access_key,
            "config_kwargs": {"s3": {"addressing_style": "path"}},
        }

        if self.endpoint_url:
            opts["client_kwargs"] = {"endpoint_url": self.endpoint_url}

        return opts

from_env() classmethod

Create S3Config from environment variables.

Environment Variables:

Endpoint (one of): - MINIO_ENDPOINT - MINIO_ENDPOINT_URL - S3_ENDPOINT_URL

Access Key (one of): - MINIO_ACCESS_KEY - AWS_ACCESS_KEY_ID

Secret Key (one of): - MINIO_SECRET_KEY - AWS_SECRET_ACCESS_KEY

Bucket Name (one of): - ASSETS_BUCKET - S3_BUCKET_NAME

Other
  • AWS_REGION (default: us-east-1)
  • USE_S3 (default: true)

Returns:

Name Type Description
S3Config S3Config

Configuration instance
Source code in src/cs_copilot/storage/config.py 
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
@classmethod
def from_env(cls) -> "S3Config":
    """
    Create S3Config from environment variables.

    Environment Variables:
    ----------------------
    Endpoint (one of):
        - MINIO_ENDPOINT
        - MINIO_ENDPOINT_URL
        - S3_ENDPOINT_URL

    Access Key (one of):
        - MINIO_ACCESS_KEY
        - AWS_ACCESS_KEY_ID

    Secret Key (one of):
        - MINIO_SECRET_KEY
        - AWS_SECRET_ACCESS_KEY

    Bucket Name (one of):
        - ASSETS_BUCKET
        - S3_BUCKET_NAME

    Other:
        - AWS_REGION (default: us-east-1)
        - USE_S3 (default: true)

    Returns:
        S3Config: Configuration instance
    """
    # Endpoint fallbacks: MINIO_ENDPOINT, MINIO_ENDPOINT_URL, S3_ENDPOINT_URL
    endpoint = (
        os.getenv("MINIO_ENDPOINT")
        or os.getenv("MINIO_ENDPOINT_URL")
        or os.getenv("S3_ENDPOINT_URL")
        or None  # When unset, let s3fs/botocore pick AWS endpoint
    )

    # Access key fallbacks: MINIO_ACCESS_KEY, AWS_ACCESS_KEY_ID
    access_key = os.getenv("MINIO_ACCESS_KEY") or os.getenv("AWS_ACCESS_KEY_ID") or "minioadmin"

    # Secret key fallbacks: MINIO_SECRET_KEY, AWS_SECRET_ACCESS_KEY
    secret_key = (
        os.getenv("MINIO_SECRET_KEY") or os.getenv("AWS_SECRET_ACCESS_KEY") or "minioadmin"
    )

    # Bucket name fallbacks: ASSETS_BUCKET, S3_BUCKET_NAME
    bucket_name = os.getenv("ASSETS_BUCKET") or os.getenv("S3_BUCKET_NAME") or "chatbot-assets"

    # Region
    region_name = os.getenv("AWS_REGION", "us-east-1")

    # Use S3 flag
    use_s3 = os.getenv("USE_S3", "true").lower() == "true"

    return cls(
        endpoint_url=endpoint,
        access_key_id=access_key,
        secret_access_key=secret_key,
        bucket_name=bucket_name,
        region_name=region_name,
        use_s3=use_s3,
    )

to_storage_options()

Convert config to fsspec/s3fs storage options.

Returns:

Name Type Description
dict dict

Storage options for fsspec.open()
Source code in src/cs_copilot/storage/config.py 
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
def to_storage_options(self) -> dict:
    """
    Convert config to fsspec/s3fs storage options.

    Returns:
        dict: Storage options for fsspec.open()
    """
    opts = {
        "key": self.access_key_id,
        "secret": self.secret_access_key,
        "config_kwargs": {"s3": {"addressing_style": "path"}},
    }

    if self.endpoint_url:
        opts["client_kwargs"] = {"endpoint_url": self.endpoint_url}

    return opts

get_s3_config()

Get S3 configuration from environment variables.

This is evaluated at call time (not import time) so that notebooks can call load_dotenv() after importing modules and still have up-to-date credentials and endpoint settings.

Returns:

Name Type Description
S3Config S3Config

Current S3 configuration
Source code in src/cs_copilot/storage/config.py 
109
110
111
112
113
114
115
116
117
118
119
120
def get_s3_config() -> S3Config:
    """
    Get S3 configuration from environment variables.

    This is evaluated at call time (not import time) so that notebooks can
    call load_dotenv() after importing modules and still have up-to-date
    credentials and endpoint settings.

    Returns:
        S3Config: Current S3 configuration
    """
    return S3Config.from_env()

is_s3_enabled()

Check if S3 is enabled based on configuration.

Returns:

Name Type Description
bool bool

True if S3 should be used
Source code in src/cs_copilot/storage/config.py 
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
def is_s3_enabled() -> bool:
    """
    Check if S3 is enabled based on configuration.

    Returns:
        bool: True if S3 should be used
    """
    config = get_s3_config()
    return config.use_s3 and all(
        [
            config.access_key_id,
            config.secret_access_key,
            config.bucket_name,
        ]
    )

client

S3 storage client for unified file operations.

Provides transparent access to files stored in S3/MinIO or locally, with automatic session-based path management.

S3

Unified storage client for S3/MinIO and local file operations.

All file operations are scoped to the current session by default, unless an absolute S3 URL or local path is provided.

Class Attributes:

prefix : str Session-scoped prefix for all relative paths

Methods:

path(rel: str) -> str Convert a relative path to an S3 URL

open(rel: str, mode: str = "rb") Open a file for reading or writing

Examples:
Write to session-scoped S3 path

with S3.open("results.csv", "w") as f: ... f.write("data")

Read from absolute S3 URL

with S3.open("s3://bucket/key.csv", "r") as f: ... data = f.read()

Read from local file

with S3.open("/tmp/local.csv", "r") as f: ... data = f.read()

Source code in src/cs_copilot/storage/client.py 
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
class S3:
    """
    Unified storage client for S3/MinIO and local file operations.

    All file operations are scoped to the current session by default,
    unless an absolute S3 URL or local path is provided.

    Class Attributes:
    -----------------
    prefix : str
        Session-scoped prefix for all relative paths

    Methods:
    --------
    path(rel: str) -> str
        Convert a relative path to an S3 URL

    open(rel: str, mode: str = "rb")
        Open a file for reading or writing

    Examples:
    ---------
    >>> # Write to session-scoped S3 path
    >>> with S3.open("results.csv", "w") as f:
    ...     f.write("data")

    >>> # Read from absolute S3 URL
    >>> with S3.open("s3://bucket/key.csv", "r") as f:
    ...     data = f.read()

    >>> # Read from local file
    >>> with S3.open("/tmp/local.csv", "r") as f:
    ...     data = f.read()
    """

    prefix = f"sessions/{SESSION_ID}"
    logger.info(f"Initialized S3 client with SESSION_ID: {SESSION_ID}")

    @classmethod
    def path(cls, rel: str) -> str:
        """
        Convert a relative path to an S3 URL.

        If an absolute S3 URL is provided (starts with s3://),
        it is returned unchanged.

        Args:
            rel: Relative path or absolute S3 URL

        Returns:
            str: Full S3 URL

        Examples:
        ---------
        >>> S3.path("data.csv")
        's3://chatbot-assets/sessions/20250121-123456-abc123/data.csv'

        >>> S3.path("s3://mybucket/data.csv")
        's3://mybucket/data.csv'
        """
        # If an absolute S3 URL is provided, pass it through unchanged
        if isinstance(rel, str) and rel.startswith("s3://"):
            return rel

        # Always produce a URL; works with fsspec & pandas
        config = get_s3_config()
        return f"s3://{config.bucket_name}/{cls.prefix}/{rel}".strip("/")

    @classmethod
    def open(cls, rel: str, mode: str = "rb"):
        """
        Open a file for reading or writing.

        Supports three types of paths:
        1. Absolute S3 URLs (s3://...)
        2. Local absolute paths (/ or file://)
        3. Relative paths (scoped to current session)

        Args:
            rel: File path (relative, absolute, or S3 URL)
            mode: File mode ('r', 'w', 'rb', 'wb', etc.)

        Returns:
            File-like object opened with the specified mode

        Examples:
        ---------
        >>> # Read from session-scoped S3 path
        >>> with S3.open("data.csv", "r") as f:
        ...     df = pd.read_csv(f)

        >>> # Write to session-scoped S3 path
        >>> with S3.open("output.csv", "w") as f:
        ...     df.to_csv(f)

        >>> # Read from absolute S3 URL
        >>> with S3.open("s3://bucket/data.csv", "r") as f:
        ...     df = pd.read_csv(f)

        >>> # Read from local file
        >>> with S3.open("/tmp/data.csv", "r") as f:
        ...     df = pd.read_csv(f)
        """
        config = get_s3_config()
        use_s3 = is_s3_enabled()

        # 1) Absolute S3 URL → open as-is
        if isinstance(rel, str) and rel.startswith("s3://"):
            return fsspec.open(rel, mode=mode, **config.to_storage_options())

        # 2) Explicit local files (absolute paths or file://) → always open locally
        if isinstance(rel, str) and (rel.startswith("/") or rel.startswith("file://")):
            if rel.startswith("file://"):
                return fsspec.open(rel, mode=mode)
            return builtins.open(rel, mode)

        # 3) If S3 is disabled, allow relative local file access
        if not use_s3:
            if isinstance(rel, str) and rel.startswith("file://"):
                return fsspec.open(rel, mode=mode)
            return builtins.open(rel, mode)

        # 4) Otherwise treat as a key relative to the session prefix and force S3
        return fsspec.open(cls.path(rel), mode=mode, **config.to_storage_options())
path(rel) classmethod

Convert a relative path to an S3 URL.

If an absolute S3 URL is provided (starts with s3://), it is returned unchanged.

Parameters:

Name Type Description Default
rel str

Relative path or absolute S3 URL

 required

Returns:

Name Type Description
str str

Full S3 URL
Examples:

S3.path("data.csv") 's3://chatbot-assets/sessions/20250121-123456-abc123/data.csv'

S3.path("s3://mybucket/data.csv") 's3://mybucket/data.csv'

Source code in src/cs_copilot/storage/client.py 
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
@classmethod
def path(cls, rel: str) -> str:
    """
    Convert a relative path to an S3 URL.

    If an absolute S3 URL is provided (starts with s3://),
    it is returned unchanged.

    Args:
        rel: Relative path or absolute S3 URL

    Returns:
        str: Full S3 URL

    Examples:
    ---------
    >>> S3.path("data.csv")
    's3://chatbot-assets/sessions/20250121-123456-abc123/data.csv'

    >>> S3.path("s3://mybucket/data.csv")
    's3://mybucket/data.csv'
    """
    # If an absolute S3 URL is provided, pass it through unchanged
    if isinstance(rel, str) and rel.startswith("s3://"):
        return rel

    # Always produce a URL; works with fsspec & pandas
    config = get_s3_config()
    return f"s3://{config.bucket_name}/{cls.prefix}/{rel}".strip("/")
open(rel, mode='rb') classmethod

Open a file for reading or writing.

Supports three types of paths: 1. Absolute S3 URLs (s3://...) 2. Local absolute paths (/ or file://) 3. Relative paths (scoped to current session)

Parameters:

Name Type Description Default
rel str

File path (relative, absolute, or S3 URL)

 required
mode str

File mode ('r', 'w', 'rb', 'wb', etc.)

 'rb'

Returns:

Type Description

File-like object opened with the specified mode
Examples:
Read from session-scoped S3 path

with S3.open("data.csv", "r") as f: ... df = pd.read_csv(f)

Write to session-scoped S3 path

with S3.open("output.csv", "w") as f: ... df.to_csv(f)

Read from absolute S3 URL

with S3.open("s3://bucket/data.csv", "r") as f: ... df = pd.read_csv(f)

Read from local file

with S3.open("/tmp/data.csv", "r") as f: ... df = pd.read_csv(f)

Source code in src/cs_copilot/storage/client.py 
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
@classmethod
def open(cls, rel: str, mode: str = "rb"):
    """
    Open a file for reading or writing.

    Supports three types of paths:
    1. Absolute S3 URLs (s3://...)
    2. Local absolute paths (/ or file://)
    3. Relative paths (scoped to current session)

    Args:
        rel: File path (relative, absolute, or S3 URL)
        mode: File mode ('r', 'w', 'rb', 'wb', etc.)

    Returns:
        File-like object opened with the specified mode

    Examples:
    ---------
    >>> # Read from session-scoped S3 path
    >>> with S3.open("data.csv", "r") as f:
    ...     df = pd.read_csv(f)

    >>> # Write to session-scoped S3 path
    >>> with S3.open("output.csv", "w") as f:
    ...     df.to_csv(f)

    >>> # Read from absolute S3 URL
    >>> with S3.open("s3://bucket/data.csv", "r") as f:
    ...     df = pd.read_csv(f)

    >>> # Read from local file
    >>> with S3.open("/tmp/data.csv", "r") as f:
    ...     df = pd.read_csv(f)
    """
    config = get_s3_config()
    use_s3 = is_s3_enabled()

    # 1) Absolute S3 URL → open as-is
    if isinstance(rel, str) and rel.startswith("s3://"):
        return fsspec.open(rel, mode=mode, **config.to_storage_options())

    # 2) Explicit local files (absolute paths or file://) → always open locally
    if isinstance(rel, str) and (rel.startswith("/") or rel.startswith("file://")):
        if rel.startswith("file://"):
            return fsspec.open(rel, mode=mode)
        return builtins.open(rel, mode)

    # 3) If S3 is disabled, allow relative local file access
    if not use_s3:
        if isinstance(rel, str) and rel.startswith("file://"):
            return fsspec.open(rel, mode=mode)
        return builtins.open(rel, mode)

    # 4) Otherwise treat as a key relative to the session prefix and force S3
    return fsspec.open(cls.path(rel), mode=mode, **config.to_storage_options())

config

Storage configuration module.

Manages S3/MinIO connection settings and environment variable fallbacks.

S3Config dataclass

Configuration for S3/MinIO storage.

Source code in src/cs_copilot/storage/config.py 
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
@dataclass
class S3Config:
    """Configuration for S3/MinIO storage."""

    endpoint_url: Optional[str]
    access_key_id: str
    secret_access_key: str
    bucket_name: str
    region_name: str = "us-east-1"
    use_s3: bool = True

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

        Environment Variables:
        ----------------------
        Endpoint (one of):
            - MINIO_ENDPOINT
            - MINIO_ENDPOINT_URL
            - S3_ENDPOINT_URL

        Access Key (one of):
            - MINIO_ACCESS_KEY
            - AWS_ACCESS_KEY_ID

        Secret Key (one of):
            - MINIO_SECRET_KEY
            - AWS_SECRET_ACCESS_KEY

        Bucket Name (one of):
            - ASSETS_BUCKET
            - S3_BUCKET_NAME

        Other:
            - AWS_REGION (default: us-east-1)
            - USE_S3 (default: true)

        Returns:
            S3Config: Configuration instance
        """
        # Endpoint fallbacks: MINIO_ENDPOINT, MINIO_ENDPOINT_URL, S3_ENDPOINT_URL
        endpoint = (
            os.getenv("MINIO_ENDPOINT")
            or os.getenv("MINIO_ENDPOINT_URL")
            or os.getenv("S3_ENDPOINT_URL")
            or None  # When unset, let s3fs/botocore pick AWS endpoint
        )

        # Access key fallbacks: MINIO_ACCESS_KEY, AWS_ACCESS_KEY_ID
        access_key = os.getenv("MINIO_ACCESS_KEY") or os.getenv("AWS_ACCESS_KEY_ID") or "minioadmin"

        # Secret key fallbacks: MINIO_SECRET_KEY, AWS_SECRET_ACCESS_KEY
        secret_key = (
            os.getenv("MINIO_SECRET_KEY") or os.getenv("AWS_SECRET_ACCESS_KEY") or "minioadmin"
        )

        # Bucket name fallbacks: ASSETS_BUCKET, S3_BUCKET_NAME
        bucket_name = os.getenv("ASSETS_BUCKET") or os.getenv("S3_BUCKET_NAME") or "chatbot-assets"

        # Region
        region_name = os.getenv("AWS_REGION", "us-east-1")

        # Use S3 flag
        use_s3 = os.getenv("USE_S3", "true").lower() == "true"

        return cls(
            endpoint_url=endpoint,
            access_key_id=access_key,
            secret_access_key=secret_key,
            bucket_name=bucket_name,
            region_name=region_name,
            use_s3=use_s3,
        )

    def to_storage_options(self) -> dict:
        """
        Convert config to fsspec/s3fs storage options.

        Returns:
            dict: Storage options for fsspec.open()
        """
        opts = {
            "key": self.access_key_id,
            "secret": self.secret_access_key,
            "config_kwargs": {"s3": {"addressing_style": "path"}},
        }

        if self.endpoint_url:
            opts["client_kwargs"] = {"endpoint_url": self.endpoint_url}

        return opts
from_env() classmethod

Create S3Config from environment variables.

Environment Variables:

Endpoint (one of): - MINIO_ENDPOINT - MINIO_ENDPOINT_URL - S3_ENDPOINT_URL

Access Key (one of): - MINIO_ACCESS_KEY - AWS_ACCESS_KEY_ID

Secret Key (one of): - MINIO_SECRET_KEY - AWS_SECRET_ACCESS_KEY

Bucket Name (one of): - ASSETS_BUCKET - S3_BUCKET_NAME

Other
  • AWS_REGION (default: us-east-1)
  • USE_S3 (default: true)

Returns:

Name Type Description
S3Config S3Config

Configuration instance
Source code in src/cs_copilot/storage/config.py 
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
@classmethod
def from_env(cls) -> "S3Config":
    """
    Create S3Config from environment variables.

    Environment Variables:
    ----------------------
    Endpoint (one of):
        - MINIO_ENDPOINT
        - MINIO_ENDPOINT_URL
        - S3_ENDPOINT_URL

    Access Key (one of):
        - MINIO_ACCESS_KEY
        - AWS_ACCESS_KEY_ID

    Secret Key (one of):
        - MINIO_SECRET_KEY
        - AWS_SECRET_ACCESS_KEY

    Bucket Name (one of):
        - ASSETS_BUCKET
        - S3_BUCKET_NAME

    Other:
        - AWS_REGION (default: us-east-1)
        - USE_S3 (default: true)

    Returns:
        S3Config: Configuration instance
    """
    # Endpoint fallbacks: MINIO_ENDPOINT, MINIO_ENDPOINT_URL, S3_ENDPOINT_URL
    endpoint = (
        os.getenv("MINIO_ENDPOINT")
        or os.getenv("MINIO_ENDPOINT_URL")
        or os.getenv("S3_ENDPOINT_URL")
        or None  # When unset, let s3fs/botocore pick AWS endpoint
    )

    # Access key fallbacks: MINIO_ACCESS_KEY, AWS_ACCESS_KEY_ID
    access_key = os.getenv("MINIO_ACCESS_KEY") or os.getenv("AWS_ACCESS_KEY_ID") or "minioadmin"

    # Secret key fallbacks: MINIO_SECRET_KEY, AWS_SECRET_ACCESS_KEY
    secret_key = (
        os.getenv("MINIO_SECRET_KEY") or os.getenv("AWS_SECRET_ACCESS_KEY") or "minioadmin"
    )

    # Bucket name fallbacks: ASSETS_BUCKET, S3_BUCKET_NAME
    bucket_name = os.getenv("ASSETS_BUCKET") or os.getenv("S3_BUCKET_NAME") or "chatbot-assets"

    # Region
    region_name = os.getenv("AWS_REGION", "us-east-1")

    # Use S3 flag
    use_s3 = os.getenv("USE_S3", "true").lower() == "true"

    return cls(
        endpoint_url=endpoint,
        access_key_id=access_key,
        secret_access_key=secret_key,
        bucket_name=bucket_name,
        region_name=region_name,
        use_s3=use_s3,
    )
to_storage_options()

Convert config to fsspec/s3fs storage options.

Returns:

Name Type Description
dict dict

Storage options for fsspec.open()
Source code in src/cs_copilot/storage/config.py 
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
def to_storage_options(self) -> dict:
    """
    Convert config to fsspec/s3fs storage options.

    Returns:
        dict: Storage options for fsspec.open()
    """
    opts = {
        "key": self.access_key_id,
        "secret": self.secret_access_key,
        "config_kwargs": {"s3": {"addressing_style": "path"}},
    }

    if self.endpoint_url:
        opts["client_kwargs"] = {"endpoint_url": self.endpoint_url}

    return opts

get_s3_config()

Get S3 configuration from environment variables.

This is evaluated at call time (not import time) so that notebooks can call load_dotenv() after importing modules and still have up-to-date credentials and endpoint settings.

Returns:

Name Type Description
S3Config S3Config

Current S3 configuration
Source code in src/cs_copilot/storage/config.py 
109
110
111
112
113
114
115
116
117
118
119
120
def get_s3_config() -> S3Config:
    """
    Get S3 configuration from environment variables.

    This is evaluated at call time (not import time) so that notebooks can
    call load_dotenv() after importing modules and still have up-to-date
    credentials and endpoint settings.

    Returns:
        S3Config: Current S3 configuration
    """
    return S3Config.from_env()

is_s3_enabled()

Check if S3 is enabled based on configuration.

Returns:

Name Type Description
bool bool

True if S3 should be used
Source code in src/cs_copilot/storage/config.py 
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
def is_s3_enabled() -> bool:
    """
    Check if S3 is enabled based on configuration.

    Returns:
        bool: True if S3 should be used
    """
    config = get_s3_config()
    return config.use_s3 and all(
        [
            config.access_key_id,
            config.secret_access_key,
            config.bucket_name,
        ]
    )