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 
 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
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
class S3(metaclass=_S3Meta):
    """
    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()
    """

    _fallback_prefix = _DEFAULT_PREFIX
    logger.info(f"Initialized S3 client with SESSION_ID: {SESSION_ID}")

    @classmethod
    def set_session_prefix(cls, prefix: str) -> None:
        """Set the storage prefix for the current execution context.

        Chainlit serves multiple chat sessions in one Python process.  A plain
        class-level prefix is shared across concurrent chats, so relative
        storage paths can leak into another user's session.  The context-local
        prefix isolates async tasks while keeping ``S3.prefix`` as a fallback
        for CLI/tests/backwards compatibility.
        """
        normalized = str(prefix).strip("/")
        if not normalized:
            raise ValueError("prefix cannot be empty")
        cls._fallback_prefix = normalized
        _SESSION_PREFIX.set(normalized)

    @classmethod
    def current_prefix(cls) -> str:
        """Return the active context-local prefix, falling back to ``S3.prefix``."""
        return _SESSION_PREFIX.get() or cls._fallback_prefix

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

        Explicit S3 and local paths are returned unchanged. Relative paths are
        resolved against the active backend.

        Args:
            rel: Relative path or absolute S3 URL

        Returns:
            str: Full S3 URL or local path

        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 cls._is_s3_url(rel) or cls._is_explicit_local_path(rel):
            return rel

        config = get_s3_config()
        if not is_s3_enabled():
            return os.fspath(cls._local_session_path(rel))

        key = PurePosixPath(cls.current_prefix().strip("/")) / str(rel).lstrip("/")
        return f"s3://{config.bucket_name}/{key.as_posix()}"

    @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()

        if cls._is_s3_url(rel):
            return fsspec.open(rel, mode=mode, **config.to_storage_options())

        if cls._is_explicit_local_path(rel):
            if rel.startswith("file://"):
                return fsspec.open(rel, mode=mode)
            return cls._open_local_path(Path(rel), mode)

        if not is_s3_enabled():
            return cls._open_local_path(cls._local_session_path(rel), mode)

        return fsspec.open(cls.path(rel), mode=mode, **config.to_storage_options())

    @staticmethod
    def _is_s3_url(path: str) -> bool:
        return isinstance(path, str) and path.startswith("s3://")

    @staticmethod
    def _is_explicit_local_path(path: str) -> bool:
        return isinstance(path, str) and (path.startswith("/") or path.startswith("file://"))

    @classmethod
    def _local_session_path(cls, rel: str) -> Path:
        return LOCAL_STORAGE_ROOT / Path(cls.current_prefix().strip("/")) / Path(rel)

    @staticmethod
    def _is_write_mode(mode: str) -> bool:
        return any(flag in mode for flag in ("w", "a", "x", "+"))

    @classmethod
    def _open_local_path(cls, path: Path, mode: str):
        if cls._is_write_mode(mode):
            path.parent.mkdir(parents=True, exist_ok=True)
        return builtins.open(path, mode)

set_session_prefix(prefix) classmethod

Set the storage prefix for the current execution context.

Chainlit serves multiple chat sessions in one Python process. A plain class-level prefix is shared across concurrent chats, so relative storage paths can leak into another user's session. The context-local prefix isolates async tasks while keeping S3.prefix as a fallback for CLI/tests/backwards compatibility.

Source code in src/cs_copilot/storage/client.py 
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
@classmethod
def set_session_prefix(cls, prefix: str) -> None:
    """Set the storage prefix for the current execution context.

    Chainlit serves multiple chat sessions in one Python process.  A plain
    class-level prefix is shared across concurrent chats, so relative
    storage paths can leak into another user's session.  The context-local
    prefix isolates async tasks while keeping ``S3.prefix`` as a fallback
    for CLI/tests/backwards compatibility.
    """
    normalized = str(prefix).strip("/")
    if not normalized:
        raise ValueError("prefix cannot be empty")
    cls._fallback_prefix = normalized
    _SESSION_PREFIX.set(normalized)

current_prefix() classmethod

Return the active context-local prefix, falling back to S3.prefix.

Source code in src/cs_copilot/storage/client.py 
107
108
109
110
@classmethod
def current_prefix(cls) -> str:
    """Return the active context-local prefix, falling back to ``S3.prefix``."""
    return _SESSION_PREFIX.get() or cls._fallback_prefix

path(rel) classmethod

Convert a relative path to an S3 URL or local session path.

Explicit S3 and local paths are returned unchanged. Relative paths are resolved against the active backend.

Parameters:

Name Type Description Default
rel str

Relative path or absolute S3 URL

 required

Returns:

Name Type Description
str str

Full S3 URL or local path
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 
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
@classmethod
def path(cls, rel: str) -> str:
    """
    Convert a relative path to an S3 URL or local session path.

    Explicit S3 and local paths are returned unchanged. Relative paths are
    resolved against the active backend.

    Args:
        rel: Relative path or absolute S3 URL

    Returns:
        str: Full S3 URL or local path

    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 cls._is_s3_url(rel) or cls._is_explicit_local_path(rel):
        return rel

    config = get_s3_config()
    if not is_s3_enabled():
        return os.fspath(cls._local_session_path(rel))

    key = PurePosixPath(cls.current_prefix().strip("/")) / str(rel).lstrip("/")
    return f"s3://{config.bucket_name}/{key.as_posix()}"

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 
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
@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()

    if cls._is_s3_url(rel):
        return fsspec.open(rel, mode=mode, **config.to_storage_options())

    if cls._is_explicit_local_path(rel):
        if rel.startswith("file://"):
            return fsspec.open(rel, mode=mode)
        return cls._open_local_path(Path(rel), mode)

    if not is_s3_enabled():
        return cls._open_local_path(cls._local_session_path(rel), mode)

    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 
 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
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
158
159
160
161
162
163
164
165
@dataclass
class S3Config:
    """Configuration for S3/MinIO storage."""

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

    @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: false)

        Returns:
            S3Config: Configuration instance
        """
        endpoint = _getenv_nonempty("MINIO_ENDPOINT", "MINIO_ENDPOINT_URL", "S3_ENDPOINT_URL")

        minio_access_key = _getenv_nonempty("MINIO_ACCESS_KEY")
        minio_secret_key = _getenv_nonempty("MINIO_SECRET_KEY")
        aws_access_key = _getenv_nonempty("AWS_ACCESS_KEY_ID")
        aws_secret_key = _getenv_nonempty("AWS_SECRET_ACCESS_KEY")

        if endpoint:
            access_key = minio_access_key or aws_access_key
            secret_key = minio_secret_key or aws_secret_key
        else:
            access_key = aws_access_key
            secret_key = aws_secret_key

        bucket_name = _getenv_nonempty("ASSETS_BUCKET", "S3_BUCKET_NAME")
        region_name = os.getenv("AWS_REGION", "us-east-1")
        use_s3 = os.getenv("USE_S3", "false").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 storage_backend(self) -> str:
        """
        Resolve the active storage backend.

        Returns:
            str: "local", "aws", or "s3-compatible"

        Raises:
            StorageConfigError: If S3 is explicitly enabled but missing required settings.
        """
        if not self.use_s3:
            return "local"

        missing_bucket = not self.bucket_name
        missing_key = not self.access_key_id
        missing_secret = not self.secret_access_key

        if self.endpoint_url:
            missing = []
            if missing_bucket:
                missing.append("ASSETS_BUCKET or S3_BUCKET_NAME")
            if missing_key:
                missing.append("MINIO_ACCESS_KEY or AWS_ACCESS_KEY_ID")
            if missing_secret:
                missing.append("MINIO_SECRET_KEY or AWS_SECRET_ACCESS_KEY")
            if missing:
                raise StorageConfigError(
                    "USE_S3=true with an S3-compatible endpoint requires "
                    + ", ".join(missing)
                    + "."
                )
            return "s3-compatible"

        missing = []
        if missing_bucket:
            missing.append("ASSETS_BUCKET or S3_BUCKET_NAME")
        if missing_key:
            missing.append("AWS_ACCESS_KEY_ID")
        if missing_secret:
            missing.append("AWS_SECRET_ACCESS_KEY")
        if missing:
            raise StorageConfigError(
                "USE_S3=true without an endpoint requires " + ", ".join(missing) + " for AWS S3."
            )
        return "aws"

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

        Returns:
            dict: Storage options for fsspec.open()
        """
        opts = {"config_kwargs": {"s3": {"addressing_style": "path"}}}

        if self.access_key_id:
            opts["key"] = self.access_key_id

        if self.secret_access_key:
            opts["secret"] = self.secret_access_key

        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: false)

Returns:

Name Type Description
S3Config S3Config

Configuration instance
Source code in src/cs_copilot/storage/config.py 
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
@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: false)

    Returns:
        S3Config: Configuration instance
    """
    endpoint = _getenv_nonempty("MINIO_ENDPOINT", "MINIO_ENDPOINT_URL", "S3_ENDPOINT_URL")

    minio_access_key = _getenv_nonempty("MINIO_ACCESS_KEY")
    minio_secret_key = _getenv_nonempty("MINIO_SECRET_KEY")
    aws_access_key = _getenv_nonempty("AWS_ACCESS_KEY_ID")
    aws_secret_key = _getenv_nonempty("AWS_SECRET_ACCESS_KEY")

    if endpoint:
        access_key = minio_access_key or aws_access_key
        secret_key = minio_secret_key or aws_secret_key
    else:
        access_key = aws_access_key
        secret_key = aws_secret_key

    bucket_name = _getenv_nonempty("ASSETS_BUCKET", "S3_BUCKET_NAME")
    region_name = os.getenv("AWS_REGION", "us-east-1")
    use_s3 = os.getenv("USE_S3", "false").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,
    )

storage_backend()

Resolve the active storage backend.

Returns:

Name Type Description
str str

"local", "aws", or "s3-compatible"

Raises:

Type Description
StorageConfigError

If S3 is explicitly enabled but missing required settings.
Source code in src/cs_copilot/storage/config.py 
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
def storage_backend(self) -> str:
    """
    Resolve the active storage backend.

    Returns:
        str: "local", "aws", or "s3-compatible"

    Raises:
        StorageConfigError: If S3 is explicitly enabled but missing required settings.
    """
    if not self.use_s3:
        return "local"

    missing_bucket = not self.bucket_name
    missing_key = not self.access_key_id
    missing_secret = not self.secret_access_key

    if self.endpoint_url:
        missing = []
        if missing_bucket:
            missing.append("ASSETS_BUCKET or S3_BUCKET_NAME")
        if missing_key:
            missing.append("MINIO_ACCESS_KEY or AWS_ACCESS_KEY_ID")
        if missing_secret:
            missing.append("MINIO_SECRET_KEY or AWS_SECRET_ACCESS_KEY")
        if missing:
            raise StorageConfigError(
                "USE_S3=true with an S3-compatible endpoint requires "
                + ", ".join(missing)
                + "."
            )
        return "s3-compatible"

    missing = []
    if missing_bucket:
        missing.append("ASSETS_BUCKET or S3_BUCKET_NAME")
    if missing_key:
        missing.append("AWS_ACCESS_KEY_ID")
    if missing_secret:
        missing.append("AWS_SECRET_ACCESS_KEY")
    if missing:
        raise StorageConfigError(
            "USE_S3=true without an endpoint requires " + ", ".join(missing) + " for AWS S3."
        )
    return "aws"

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 
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
def to_storage_options(self) -> dict:
    """
    Convert config to fsspec/s3fs storage options.

    Returns:
        dict: Storage options for fsspec.open()
    """
    opts = {"config_kwargs": {"s3": {"addressing_style": "path"}}}

    if self.access_key_id:
        opts["key"] = self.access_key_id

    if self.secret_access_key:
        opts["secret"] = self.secret_access_key

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

    return opts

StorageConfigError

Bases: ValueError

Raised when S3 is explicitly enabled but the runtime config is incomplete.

Source code in src/cs_copilot/storage/config.py 
14
15
class StorageConfigError(ValueError):
    """Raised when S3 is explicitly enabled but the runtime config is incomplete."""

OutputLayout dataclass

Structured output path builder for one workflow.

Source code in src/cs_copilot/storage/layout.py 
32
33
34
35
36
37
38
39
40
41
42
43
44
45
@dataclass(frozen=True)
class OutputLayout:
    """Structured output path builder for one workflow."""

    workflow_id: str

    def rel_path(self, operation: OutputOperation, *parts: str) -> str:
        cleaned_parts = [sanitize_path_part(part) for part in parts if str(part).strip()]
        path = PurePosixPath(WORKFLOWS_DIR, self.workflow_id, operation.value, *cleaned_parts)
        return path.as_posix()

    @property
    def manifest_rel_path(self) -> str:
        return PurePosixPath(WORKFLOWS_DIR, self.workflow_id, "manifest.json").as_posix()

OutputOperation

Bases: str, Enum

Top-level operation folders inside a workflow.

Source code in src/cs_copilot/storage/layout.py 
23
24
25
26
27
28
29
class OutputOperation(str, Enum):
    """Top-level operation folders inside a workflow."""

    CHEMICAL_SPACE = "01_chemical_space"
    ANALOG_GENERATION = "02_analog_generation"
    RETROSYNTHESIS = "03_retrosynthesis"
    REPORTS = "reports"

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 
168
169
170
171
172
173
174
175
176
177
178
179
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 
182
183
184
185
186
187
188
189
def is_s3_enabled() -> bool:
    """
    Check if S3 is enabled based on configuration.

    Returns:
        bool: True if S3 should be used
    """
    return get_s3_config().storage_backend() != "local"

ensure_output_context(session_state=None, *, workflow_slug=None)

Ensure a layout context exists and return it.

One storage session gets one workflow root. The workflow id is derived from the active S3/local session prefix, so tools that do not share the same in-memory session_state still write under the same root.

Source code in src/cs_copilot/storage/layout.py 
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
def ensure_output_context(
    session_state: Optional[dict[str, Any]] = None,
    *,
    workflow_slug: Optional[str] = None,
) -> dict[str, Any]:
    """Ensure a layout context exists and return it.

    One storage session gets one workflow root.  The workflow id is derived from
    the active S3/local session prefix, so tools that do not share the same
    in-memory session_state still write under the same root.
    """

    workflow_id = _session_workflow_id()
    context = {"layout_version": LAYOUT_VERSION, "workflow_id": workflow_id}

    if isinstance(session_state, dict):
        existing = session_state.get(OUTPUT_CONTEXT_KEY)
        if (
            isinstance(existing, dict)
            and existing.get("layout_version") == LAYOUT_VERSION
            and existing.get("workflow_id") == workflow_id
        ):
            return existing
        session_state[OUTPUT_CONTEXT_KEY] = context

    return context

scoped_artifact_path(path, operation, *folders, session_state=None, workflow_slug=None)

Scope an ordinary relative artifact path into the workflow layout.

Source code in src/cs_copilot/storage/layout.py 
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
def scoped_artifact_path(
    path: str,
    operation: OutputOperation,
    *folders: str,
    session_state: Optional[dict[str, Any]] = None,
    workflow_slug: Optional[str] = None,
) -> str:
    """Scope an ordinary relative artifact path into the workflow layout."""

    if is_explicit_storage_path(path) or is_workflow_scoped_path(path):
        return path
    filename = sanitize_path_part(path)
    return operation_rel_path(
        operation,
        *folders,
        filename,
        session_state=session_state,
        workflow_slug=workflow_slug,
    )

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 
 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
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
class S3(metaclass=_S3Meta):
    """
    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()
    """

    _fallback_prefix = _DEFAULT_PREFIX
    logger.info(f"Initialized S3 client with SESSION_ID: {SESSION_ID}")

    @classmethod
    def set_session_prefix(cls, prefix: str) -> None:
        """Set the storage prefix for the current execution context.

        Chainlit serves multiple chat sessions in one Python process.  A plain
        class-level prefix is shared across concurrent chats, so relative
        storage paths can leak into another user's session.  The context-local
        prefix isolates async tasks while keeping ``S3.prefix`` as a fallback
        for CLI/tests/backwards compatibility.
        """
        normalized = str(prefix).strip("/")
        if not normalized:
            raise ValueError("prefix cannot be empty")
        cls._fallback_prefix = normalized
        _SESSION_PREFIX.set(normalized)

    @classmethod
    def current_prefix(cls) -> str:
        """Return the active context-local prefix, falling back to ``S3.prefix``."""
        return _SESSION_PREFIX.get() or cls._fallback_prefix

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

        Explicit S3 and local paths are returned unchanged. Relative paths are
        resolved against the active backend.

        Args:
            rel: Relative path or absolute S3 URL

        Returns:
            str: Full S3 URL or local path

        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 cls._is_s3_url(rel) or cls._is_explicit_local_path(rel):
            return rel

        config = get_s3_config()
        if not is_s3_enabled():
            return os.fspath(cls._local_session_path(rel))

        key = PurePosixPath(cls.current_prefix().strip("/")) / str(rel).lstrip("/")
        return f"s3://{config.bucket_name}/{key.as_posix()}"

    @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()

        if cls._is_s3_url(rel):
            return fsspec.open(rel, mode=mode, **config.to_storage_options())

        if cls._is_explicit_local_path(rel):
            if rel.startswith("file://"):
                return fsspec.open(rel, mode=mode)
            return cls._open_local_path(Path(rel), mode)

        if not is_s3_enabled():
            return cls._open_local_path(cls._local_session_path(rel), mode)

        return fsspec.open(cls.path(rel), mode=mode, **config.to_storage_options())

    @staticmethod
    def _is_s3_url(path: str) -> bool:
        return isinstance(path, str) and path.startswith("s3://")

    @staticmethod
    def _is_explicit_local_path(path: str) -> bool:
        return isinstance(path, str) and (path.startswith("/") or path.startswith("file://"))

    @classmethod
    def _local_session_path(cls, rel: str) -> Path:
        return LOCAL_STORAGE_ROOT / Path(cls.current_prefix().strip("/")) / Path(rel)

    @staticmethod
    def _is_write_mode(mode: str) -> bool:
        return any(flag in mode for flag in ("w", "a", "x", "+"))

    @classmethod
    def _open_local_path(cls, path: Path, mode: str):
        if cls._is_write_mode(mode):
            path.parent.mkdir(parents=True, exist_ok=True)
        return builtins.open(path, mode)
set_session_prefix(prefix) classmethod

Set the storage prefix for the current execution context.

Chainlit serves multiple chat sessions in one Python process. A plain class-level prefix is shared across concurrent chats, so relative storage paths can leak into another user's session. The context-local prefix isolates async tasks while keeping S3.prefix as a fallback for CLI/tests/backwards compatibility.

Source code in src/cs_copilot/storage/client.py 
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
@classmethod
def set_session_prefix(cls, prefix: str) -> None:
    """Set the storage prefix for the current execution context.

    Chainlit serves multiple chat sessions in one Python process.  A plain
    class-level prefix is shared across concurrent chats, so relative
    storage paths can leak into another user's session.  The context-local
    prefix isolates async tasks while keeping ``S3.prefix`` as a fallback
    for CLI/tests/backwards compatibility.
    """
    normalized = str(prefix).strip("/")
    if not normalized:
        raise ValueError("prefix cannot be empty")
    cls._fallback_prefix = normalized
    _SESSION_PREFIX.set(normalized)
current_prefix() classmethod

Return the active context-local prefix, falling back to S3.prefix.

Source code in src/cs_copilot/storage/client.py 
107
108
109
110
@classmethod
def current_prefix(cls) -> str:
    """Return the active context-local prefix, falling back to ``S3.prefix``."""
    return _SESSION_PREFIX.get() or cls._fallback_prefix
path(rel) classmethod

Convert a relative path to an S3 URL or local session path.

Explicit S3 and local paths are returned unchanged. Relative paths are resolved against the active backend.

Parameters:

Name Type Description Default
rel str

Relative path or absolute S3 URL

 required

Returns:

Name Type Description
str str

Full S3 URL or local path
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 
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
@classmethod
def path(cls, rel: str) -> str:
    """
    Convert a relative path to an S3 URL or local session path.

    Explicit S3 and local paths are returned unchanged. Relative paths are
    resolved against the active backend.

    Args:
        rel: Relative path or absolute S3 URL

    Returns:
        str: Full S3 URL or local path

    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 cls._is_s3_url(rel) or cls._is_explicit_local_path(rel):
        return rel

    config = get_s3_config()
    if not is_s3_enabled():
        return os.fspath(cls._local_session_path(rel))

    key = PurePosixPath(cls.current_prefix().strip("/")) / str(rel).lstrip("/")
    return f"s3://{config.bucket_name}/{key.as_posix()}"
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 
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
@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()

    if cls._is_s3_url(rel):
        return fsspec.open(rel, mode=mode, **config.to_storage_options())

    if cls._is_explicit_local_path(rel):
        if rel.startswith("file://"):
            return fsspec.open(rel, mode=mode)
        return cls._open_local_path(Path(rel), mode)

    if not is_s3_enabled():
        return cls._open_local_path(cls._local_session_path(rel), mode)

    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.

StorageConfigError

Bases: ValueError

Raised when S3 is explicitly enabled but the runtime config is incomplete.

Source code in src/cs_copilot/storage/config.py 
14
15
class StorageConfigError(ValueError):
    """Raised when S3 is explicitly enabled but the runtime config is incomplete."""

S3Config dataclass

Configuration for S3/MinIO storage.

Source code in src/cs_copilot/storage/config.py 
 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
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
158
159
160
161
162
163
164
165
@dataclass
class S3Config:
    """Configuration for S3/MinIO storage."""

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

    @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: false)

        Returns:
            S3Config: Configuration instance
        """
        endpoint = _getenv_nonempty("MINIO_ENDPOINT", "MINIO_ENDPOINT_URL", "S3_ENDPOINT_URL")

        minio_access_key = _getenv_nonempty("MINIO_ACCESS_KEY")
        minio_secret_key = _getenv_nonempty("MINIO_SECRET_KEY")
        aws_access_key = _getenv_nonempty("AWS_ACCESS_KEY_ID")
        aws_secret_key = _getenv_nonempty("AWS_SECRET_ACCESS_KEY")

        if endpoint:
            access_key = minio_access_key or aws_access_key
            secret_key = minio_secret_key or aws_secret_key
        else:
            access_key = aws_access_key
            secret_key = aws_secret_key

        bucket_name = _getenv_nonempty("ASSETS_BUCKET", "S3_BUCKET_NAME")
        region_name = os.getenv("AWS_REGION", "us-east-1")
        use_s3 = os.getenv("USE_S3", "false").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 storage_backend(self) -> str:
        """
        Resolve the active storage backend.

        Returns:
            str: "local", "aws", or "s3-compatible"

        Raises:
            StorageConfigError: If S3 is explicitly enabled but missing required settings.
        """
        if not self.use_s3:
            return "local"

        missing_bucket = not self.bucket_name
        missing_key = not self.access_key_id
        missing_secret = not self.secret_access_key

        if self.endpoint_url:
            missing = []
            if missing_bucket:
                missing.append("ASSETS_BUCKET or S3_BUCKET_NAME")
            if missing_key:
                missing.append("MINIO_ACCESS_KEY or AWS_ACCESS_KEY_ID")
            if missing_secret:
                missing.append("MINIO_SECRET_KEY or AWS_SECRET_ACCESS_KEY")
            if missing:
                raise StorageConfigError(
                    "USE_S3=true with an S3-compatible endpoint requires "
                    + ", ".join(missing)
                    + "."
                )
            return "s3-compatible"

        missing = []
        if missing_bucket:
            missing.append("ASSETS_BUCKET or S3_BUCKET_NAME")
        if missing_key:
            missing.append("AWS_ACCESS_KEY_ID")
        if missing_secret:
            missing.append("AWS_SECRET_ACCESS_KEY")
        if missing:
            raise StorageConfigError(
                "USE_S3=true without an endpoint requires " + ", ".join(missing) + " for AWS S3."
            )
        return "aws"

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

        Returns:
            dict: Storage options for fsspec.open()
        """
        opts = {"config_kwargs": {"s3": {"addressing_style": "path"}}}

        if self.access_key_id:
            opts["key"] = self.access_key_id

        if self.secret_access_key:
            opts["secret"] = self.secret_access_key

        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: false)

Returns:

Name Type Description
S3Config S3Config

Configuration instance
Source code in src/cs_copilot/storage/config.py 
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
@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: false)

    Returns:
        S3Config: Configuration instance
    """
    endpoint = _getenv_nonempty("MINIO_ENDPOINT", "MINIO_ENDPOINT_URL", "S3_ENDPOINT_URL")

    minio_access_key = _getenv_nonempty("MINIO_ACCESS_KEY")
    minio_secret_key = _getenv_nonempty("MINIO_SECRET_KEY")
    aws_access_key = _getenv_nonempty("AWS_ACCESS_KEY_ID")
    aws_secret_key = _getenv_nonempty("AWS_SECRET_ACCESS_KEY")

    if endpoint:
        access_key = minio_access_key or aws_access_key
        secret_key = minio_secret_key or aws_secret_key
    else:
        access_key = aws_access_key
        secret_key = aws_secret_key

    bucket_name = _getenv_nonempty("ASSETS_BUCKET", "S3_BUCKET_NAME")
    region_name = os.getenv("AWS_REGION", "us-east-1")
    use_s3 = os.getenv("USE_S3", "false").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,
    )
storage_backend()

Resolve the active storage backend.

Returns:

Name Type Description
str str

"local", "aws", or "s3-compatible"

Raises:

Type Description
StorageConfigError

If S3 is explicitly enabled but missing required settings.
Source code in src/cs_copilot/storage/config.py 
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
def storage_backend(self) -> str:
    """
    Resolve the active storage backend.

    Returns:
        str: "local", "aws", or "s3-compatible"

    Raises:
        StorageConfigError: If S3 is explicitly enabled but missing required settings.
    """
    if not self.use_s3:
        return "local"

    missing_bucket = not self.bucket_name
    missing_key = not self.access_key_id
    missing_secret = not self.secret_access_key

    if self.endpoint_url:
        missing = []
        if missing_bucket:
            missing.append("ASSETS_BUCKET or S3_BUCKET_NAME")
        if missing_key:
            missing.append("MINIO_ACCESS_KEY or AWS_ACCESS_KEY_ID")
        if missing_secret:
            missing.append("MINIO_SECRET_KEY or AWS_SECRET_ACCESS_KEY")
        if missing:
            raise StorageConfigError(
                "USE_S3=true with an S3-compatible endpoint requires "
                + ", ".join(missing)
                + "."
            )
        return "s3-compatible"

    missing = []
    if missing_bucket:
        missing.append("ASSETS_BUCKET or S3_BUCKET_NAME")
    if missing_key:
        missing.append("AWS_ACCESS_KEY_ID")
    if missing_secret:
        missing.append("AWS_SECRET_ACCESS_KEY")
    if missing:
        raise StorageConfigError(
            "USE_S3=true without an endpoint requires " + ", ".join(missing) + " for AWS S3."
        )
    return "aws"
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 
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
def to_storage_options(self) -> dict:
    """
    Convert config to fsspec/s3fs storage options.

    Returns:
        dict: Storage options for fsspec.open()
    """
    opts = {"config_kwargs": {"s3": {"addressing_style": "path"}}}

    if self.access_key_id:
        opts["key"] = self.access_key_id

    if self.secret_access_key:
        opts["secret"] = self.secret_access_key

    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 
168
169
170
171
172
173
174
175
176
177
178
179
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 
182
183
184
185
186
187
188
189
def is_s3_enabled() -> bool:
    """
    Check if S3 is enabled based on configuration.

    Returns:
        bool: True if S3 should be used
    """
    return get_s3_config().storage_backend() != "local"

layout

Session workflow output layout helpers.

OutputOperation

Bases: str, Enum

Top-level operation folders inside a workflow.

Source code in src/cs_copilot/storage/layout.py 
23
24
25
26
27
28
29
class OutputOperation(str, Enum):
    """Top-level operation folders inside a workflow."""

    CHEMICAL_SPACE = "01_chemical_space"
    ANALOG_GENERATION = "02_analog_generation"
    RETROSYNTHESIS = "03_retrosynthesis"
    REPORTS = "reports"

OutputLayout dataclass

Structured output path builder for one workflow.

Source code in src/cs_copilot/storage/layout.py 
32
33
34
35
36
37
38
39
40
41
42
43
44
45
@dataclass(frozen=True)
class OutputLayout:
    """Structured output path builder for one workflow."""

    workflow_id: str

    def rel_path(self, operation: OutputOperation, *parts: str) -> str:
        cleaned_parts = [sanitize_path_part(part) for part in parts if str(part).strip()]
        path = PurePosixPath(WORKFLOWS_DIR, self.workflow_id, operation.value, *cleaned_parts)
        return path.as_posix()

    @property
    def manifest_rel_path(self) -> str:
        return PurePosixPath(WORKFLOWS_DIR, self.workflow_id, "manifest.json").as_posix()

sanitize_path_part(value, *, default='artifact')

Return a safe single path component while preserving useful suffix dots.

Source code in src/cs_copilot/storage/layout.py 
48
49
50
51
52
53
54
def sanitize_path_part(value: Any, *, default: str = "artifact") -> str:
    """Return a safe single path component while preserving useful suffix dots."""

    text = str(value or "").replace("\\", "/").strip().strip("/")
    text = PurePosixPath(text).name if "/" in text else text
    text = _SAFE_PART_RE.sub("_", text).strip("._-")
    return text or default

ensure_output_context(session_state=None, *, workflow_slug=None)

Ensure a layout context exists and return it.

One storage session gets one workflow root. The workflow id is derived from the active S3/local session prefix, so tools that do not share the same in-memory session_state still write under the same root.

Source code in src/cs_copilot/storage/layout.py 
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
def ensure_output_context(
    session_state: Optional[dict[str, Any]] = None,
    *,
    workflow_slug: Optional[str] = None,
) -> dict[str, Any]:
    """Ensure a layout context exists and return it.

    One storage session gets one workflow root.  The workflow id is derived from
    the active S3/local session prefix, so tools that do not share the same
    in-memory session_state still write under the same root.
    """

    workflow_id = _session_workflow_id()
    context = {"layout_version": LAYOUT_VERSION, "workflow_id": workflow_id}

    if isinstance(session_state, dict):
        existing = session_state.get(OUTPUT_CONTEXT_KEY)
        if (
            isinstance(existing, dict)
            and existing.get("layout_version") == LAYOUT_VERSION
            and existing.get("workflow_id") == workflow_id
        ):
            return existing
        session_state[OUTPUT_CONTEXT_KEY] = context

    return context

scoped_artifact_path(path, operation, *folders, session_state=None, workflow_slug=None)

Scope an ordinary relative artifact path into the workflow layout.

Source code in src/cs_copilot/storage/layout.py 
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
def scoped_artifact_path(
    path: str,
    operation: OutputOperation,
    *folders: str,
    session_state: Optional[dict[str, Any]] = None,
    workflow_slug: Optional[str] = None,
) -> str:
    """Scope an ordinary relative artifact path into the workflow layout."""

    if is_explicit_storage_path(path) or is_workflow_scoped_path(path):
        return path
    filename = sanitize_path_part(path)
    return operation_rel_path(
        operation,
        *folders,
        filename,
        session_state=session_state,
        workflow_slug=workflow_slug,
    )