Skip to content

datarobot_genai.drmcputils.files.file_system_store

file_system_store

Hierarchical filesystem seam for drtools, backed by the DataRobot Files API.

Unlike the flat :class:~datarobot_genai.drmcputils.files.store.BlobStore (one blob == one Files container), this store exposes DataRobot's hierarchical filesystem: files live inside a catalog item addressed as dr://<catalog_item_id>/path/to/file. It is a thin async wrapper around :class:datarobot.fs.DataRobotFileSystem (an fsspec implementation).

Design mirrors :mod:datarobot_genai.drmcputils.files.store:

  • A minimal :class:FileSystemStore Protocol so higher layers depend on the contract, not the concrete backend (an in-memory backend can satisfy it in tests).
  • The fsspec filesystem is synchronous; its calls are dispatched to a worker thread via :func:asyncio.to_thread, which copies the current context so the per-request client configured by :func:request_user_dr_sdk stays in effect inside the thread.
  • DataRobot/fsspec errors are normalised to :class:ToolError so tool code never leaks SDK or OS exception types.

DirectoryNotEmptyError

Bases: Exception

Raised by :meth:FileSystemStore.delete for a non-recursive delete of a non-empty directory.

Distinct from the silent no-op for an already-missing path: that case has nothing to signal, while this one represents a real guard rejecting the request, which callers should surface rather than report as a successful deletion.

Source code in datarobot_genai/drmcputils/files/file_system_store.py 
74
75
76
77
78
79
80
class DirectoryNotEmptyError(Exception):
    """Raised by :meth:`FileSystemStore.delete` for a non-recursive delete of a non-empty directory.

    Distinct from the silent no-op for an already-missing path: that case has nothing to
    signal, while this one represents a real guard rejecting the request, which callers
    should surface rather than report as a successful deletion.
    """

FileEntry dataclass

A serializable description of a file or directory in the DataRobot filesystem.

Built from the fsspec FileInfo mapping the backend returns. name is the path without the dr:// protocol prefix (e.g. <catalog_id>/data/file.csv); directories carry type == "directory" and size == 0.

Source code in datarobot_genai/drmcputils/files/file_system_store.py 
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
@dataclass(frozen=True, slots=True)
class FileEntry:
    """A serializable description of a file or directory in the DataRobot filesystem.

    Built from the fsspec ``FileInfo`` mapping the backend returns. ``name`` is
    the path *without* the ``dr://`` protocol prefix (e.g.
    ``<catalog_id>/data/file.csv``); directories carry ``type == "directory"``
    and ``size == 0``.
    """

    name: str
    type: str
    size: int = 0
    format: str | None = None
    created_at: str | None = None

    @classmethod
    def from_info(cls, info: Mapping[str, Any]) -> FileEntry:
        """Build a :class:`FileEntry` from an fsspec ``FileInfo`` mapping."""
        created = info.get("created_at")
        if isinstance(created, datetime):
            created = created.isoformat()
        return cls(
            name=str(info.get("name", "")),
            type=str(info.get("type", "")),
            size=int(info.get("size") or 0),
            format=info.get("format"),
            created_at=created,
        )

    def to_dict(self) -> dict[str, Any]:
        """Return a plain JSON-serializable dict for tool responses."""
        return {
            "name": self.name,
            "type": self.type,
            "size": self.size,
            "format": self.format,
            "created_at": self.created_at,
        }

from_info classmethod 

from_info(info: Mapping[str, Any]) -> FileEntry

Build a :class:FileEntry from an fsspec FileInfo mapping.

Source code in datarobot_genai/drmcputils/files/file_system_store.py 
133
134
135
136
137
138
139
140
141
142
143
144
145
@classmethod
def from_info(cls, info: Mapping[str, Any]) -> FileEntry:
    """Build a :class:`FileEntry` from an fsspec ``FileInfo`` mapping."""
    created = info.get("created_at")
    if isinstance(created, datetime):
        created = created.isoformat()
    return cls(
        name=str(info.get("name", "")),
        type=str(info.get("type", "")),
        size=int(info.get("size") or 0),
        format=info.get("format"),
        created_at=created,
    )

to_dict 

to_dict() -> dict[str, Any]

Return a plain JSON-serializable dict for tool responses.

Source code in datarobot_genai/drmcputils/files/file_system_store.py 
147
148
149
150
151
152
153
154
155
def to_dict(self) -> dict[str, Any]:
    """Return a plain JSON-serializable dict for tool responses."""
    return {
        "name": self.name,
        "type": self.type,
        "size": self.size,
        "format": self.format,
        "created_at": self.created_at,
    }

FileSystemStore

Bases: Protocol

Read seam over the DataRobot hierarchical filesystem. Implementations must be async-safe.

Source code in datarobot_genai/drmcputils/files/file_system_store.py 
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
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
@runtime_checkable
class FileSystemStore(Protocol):
    """Read seam over the DataRobot hierarchical filesystem. Implementations must be async-safe."""

    async def ls(self, path: str, *, detail: bool = True) -> list[FileEntry]:
        """List the immediate children of ``path`` (catalog items at ``dr://`` root)."""
        ...

    async def find(
        self, path: str, *, maxdepth: int | None = None, withdirs: bool = False
    ) -> list[FileEntry]:
        """Recursively list everything under ``path`` (posix ``find`` semantics, no globs)."""
        ...

    async def glob(self, pattern: str, *, maxdepth: int | None = None) -> list[FileEntry]:
        """List entries matching a glob ``pattern`` (e.g. ``dr://<id>/**/*.csv``)."""
        ...

    async def tree(
        self,
        path: str,
        *,
        recursion_limit: int = 2,
        max_display: int = 25,
        display_size: bool = False,
    ) -> str:
        """Return a compact, indented tree view of ``path``."""
        ...

    async def info(self, path: str) -> FileEntry:
        """Return metadata for a single file or directory ``path``."""
        ...

    async def read(self, path: str, *, start: int | None = None, end: int | None = None) -> bytes:
        """Read (a byte range of) the file at ``path``."""
        ...

    async def sign(self, path: str, *, expiration: int = DEFAULT_SIGN_EXPIRATION_SECONDS) -> str:
        """Return a temporary signed download URL for the file at ``path``."""
        ...

    async def write(self, path: str, data: bytes, *, mode: str = "overwrite") -> None:
        """Write ``data`` to the file at ``path`` ('overwrite' or 'create' if absent)."""
        ...

    async def upload(
        self,
        local_path: str | list[str],
        dest: str | list[str],
        *,
        recursive: bool = False,
        maxdepth: int | None = None,
        overwrite: OverwriteStrategyName = "rename",
    ) -> None:
        """Upload local file(s)/director(ies) into the filesystem.

        Content is streamed in chunks (no inline size cap) and multiple files are
        batched into a single optimized request when possible.
        """
        ...

    async def create_dir(self) -> str:
        """Create an empty catalog item directory and return its id."""
        ...

    async def delete(
        self, path: str, *, recursive: bool = False, maxdepth: int | None = None
    ) -> None:
        """Delete file(s) or director(ies) at ``path`` (silent if absent).

        Raises :class:`DirectoryNotEmptyError` if ``recursive`` is ``False`` and ``path`` is
        an existing, non-empty directory (nothing is deleted in that case).
        """
        ...

    async def copy(
        self,
        source: str,
        dest: str,
        *,
        recursive: bool = False,
        maxdepth: int | None = None,
        overwrite: OverwriteStrategyName = "rename",
    ) -> None:
        """Copy ``source`` to ``dest`` within the filesystem."""
        ...

    async def move(
        self,
        source: str,
        dest: str,
        *,
        recursive: bool = False,
        maxdepth: int | None = None,
        overwrite: OverwriteStrategyName = "rename",
    ) -> None:
        """Move/rename ``source`` to ``dest`` within the filesystem."""
        ...

    async def clone(self, path_or_id: str, *, files_to_omit: list[str] | None = None) -> str:
        """Clone a catalog item directory and return the new catalog item id."""
        ...

    async def import_from_url(
        self,
        path: str,
        url: str,
        *,
        unpack_archive: bool = True,
        overwrite: OverwriteStrategyName = "rename",
    ) -> str:
        """Start a URL import and return the async status id."""
        ...

    async def import_from_data_source(
        self,
        path: str,
        data_source_id: str,
        *,
        credential_id: str | None = None,
        unpack_archive: bool = True,
        overwrite: OverwriteStrategyName = "rename",
    ) -> str:
        """Start a data-source import and return the async status id."""
        ...

    async def get_status(self, status_id: str) -> dict[str, Any]:
        """Fetch the current async status payload for a single import job."""
        ...

ls async 

ls(path: str, *, detail: bool = True) -> list[FileEntry]

List the immediate children of path (catalog items at dr:// root).

Source code in datarobot_genai/drmcputils/files/file_system_store.py 
162
163
164
async def ls(self, path: str, *, detail: bool = True) -> list[FileEntry]:
    """List the immediate children of ``path`` (catalog items at ``dr://`` root)."""
    ...

find async 

find(path: str, *, maxdepth: int | None = None, withdirs: bool = False) -> list[FileEntry]

Recursively list everything under path (posix find semantics, no globs).

Source code in datarobot_genai/drmcputils/files/file_system_store.py 
166
167
168
169
170
async def find(
    self, path: str, *, maxdepth: int | None = None, withdirs: bool = False
) -> list[FileEntry]:
    """Recursively list everything under ``path`` (posix ``find`` semantics, no globs)."""
    ...

glob async 

glob(pattern: str, *, maxdepth: int | None = None) -> list[FileEntry]

List entries matching a glob pattern (e.g. dr://<id>/**/*.csv).

Source code in datarobot_genai/drmcputils/files/file_system_store.py 
172
173
174
async def glob(self, pattern: str, *, maxdepth: int | None = None) -> list[FileEntry]:
    """List entries matching a glob ``pattern`` (e.g. ``dr://<id>/**/*.csv``)."""
    ...

tree async 

tree(path: str, *, recursion_limit: int = 2, max_display: int = 25, display_size: bool = False) -> str

Return a compact, indented tree view of path.

Source code in datarobot_genai/drmcputils/files/file_system_store.py 
176
177
178
179
180
181
182
183
184
185
async def tree(
    self,
    path: str,
    *,
    recursion_limit: int = 2,
    max_display: int = 25,
    display_size: bool = False,
) -> str:
    """Return a compact, indented tree view of ``path``."""
    ...

info async 

info(path: str) -> FileEntry

Return metadata for a single file or directory path.

Source code in datarobot_genai/drmcputils/files/file_system_store.py 
187
188
189
async def info(self, path: str) -> FileEntry:
    """Return metadata for a single file or directory ``path``."""
    ...

read async 

read(path: str, *, start: int | None = None, end: int | None = None) -> bytes

Read (a byte range of) the file at path.

Source code in datarobot_genai/drmcputils/files/file_system_store.py 
191
192
193
async def read(self, path: str, *, start: int | None = None, end: int | None = None) -> bytes:
    """Read (a byte range of) the file at ``path``."""
    ...

sign async 

sign(path: str, *, expiration: int = DEFAULT_SIGN_EXPIRATION_SECONDS) -> str

Return a temporary signed download URL for the file at path.

Source code in datarobot_genai/drmcputils/files/file_system_store.py 
195
196
197
async def sign(self, path: str, *, expiration: int = DEFAULT_SIGN_EXPIRATION_SECONDS) -> str:
    """Return a temporary signed download URL for the file at ``path``."""
    ...

write async 

write(path: str, data: bytes, *, mode: str = 'overwrite') -> None

Write data to the file at path ('overwrite' or 'create' if absent).

Source code in datarobot_genai/drmcputils/files/file_system_store.py 
199
200
201
async def write(self, path: str, data: bytes, *, mode: str = "overwrite") -> None:
    """Write ``data`` to the file at ``path`` ('overwrite' or 'create' if absent)."""
    ...

upload async 

upload(local_path: str | list[str], dest: str | list[str], *, recursive: bool = False, maxdepth: int | None = None, overwrite: OverwriteStrategyName = 'rename') -> None

Upload local file(s)/director(ies) into the filesystem.

Content is streamed in chunks (no inline size cap) and multiple files are batched into a single optimized request when possible.

Source code in datarobot_genai/drmcputils/files/file_system_store.py 
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
async def upload(
    self,
    local_path: str | list[str],
    dest: str | list[str],
    *,
    recursive: bool = False,
    maxdepth: int | None = None,
    overwrite: OverwriteStrategyName = "rename",
) -> None:
    """Upload local file(s)/director(ies) into the filesystem.

    Content is streamed in chunks (no inline size cap) and multiple files are
    batched into a single optimized request when possible.
    """
    ...

create_dir async 

create_dir() -> str

Create an empty catalog item directory and return its id.

Source code in datarobot_genai/drmcputils/files/file_system_store.py 
219
220
221
async def create_dir(self) -> str:
    """Create an empty catalog item directory and return its id."""
    ...

delete async 

delete(path: str, *, recursive: bool = False, maxdepth: int | None = None) -> None

Delete file(s) or director(ies) at path (silent if absent).

Raises :class:DirectoryNotEmptyError if recursive is False and path is an existing, non-empty directory (nothing is deleted in that case).

Source code in datarobot_genai/drmcputils/files/file_system_store.py 
223
224
225
226
227
228
229
230
231
async def delete(
    self, path: str, *, recursive: bool = False, maxdepth: int | None = None
) -> None:
    """Delete file(s) or director(ies) at ``path`` (silent if absent).

    Raises :class:`DirectoryNotEmptyError` if ``recursive`` is ``False`` and ``path`` is
    an existing, non-empty directory (nothing is deleted in that case).
    """
    ...

copy async 

copy(source: str, dest: str, *, recursive: bool = False, maxdepth: int | None = None, overwrite: OverwriteStrategyName = 'rename') -> None

Copy source to dest within the filesystem.

Source code in datarobot_genai/drmcputils/files/file_system_store.py 
233
234
235
236
237
238
239
240
241
242
243
async def copy(
    self,
    source: str,
    dest: str,
    *,
    recursive: bool = False,
    maxdepth: int | None = None,
    overwrite: OverwriteStrategyName = "rename",
) -> None:
    """Copy ``source`` to ``dest`` within the filesystem."""
    ...

move async 

move(source: str, dest: str, *, recursive: bool = False, maxdepth: int | None = None, overwrite: OverwriteStrategyName = 'rename') -> None

Move/rename source to dest within the filesystem.

Source code in datarobot_genai/drmcputils/files/file_system_store.py 
245
246
247
248
249
250
251
252
253
254
255
async def move(
    self,
    source: str,
    dest: str,
    *,
    recursive: bool = False,
    maxdepth: int | None = None,
    overwrite: OverwriteStrategyName = "rename",
) -> None:
    """Move/rename ``source`` to ``dest`` within the filesystem."""
    ...

clone async 

clone(path_or_id: str, *, files_to_omit: list[str] | None = None) -> str

Clone a catalog item directory and return the new catalog item id.

Source code in datarobot_genai/drmcputils/files/file_system_store.py 
257
258
259
async def clone(self, path_or_id: str, *, files_to_omit: list[str] | None = None) -> str:
    """Clone a catalog item directory and return the new catalog item id."""
    ...

import_from_url async 

import_from_url(path: str, url: str, *, unpack_archive: bool = True, overwrite: OverwriteStrategyName = 'rename') -> str

Start a URL import and return the async status id.

Source code in datarobot_genai/drmcputils/files/file_system_store.py 
261
262
263
264
265
266
267
268
269
270
async def import_from_url(
    self,
    path: str,
    url: str,
    *,
    unpack_archive: bool = True,
    overwrite: OverwriteStrategyName = "rename",
) -> str:
    """Start a URL import and return the async status id."""
    ...

import_from_data_source async 

import_from_data_source(path: str, data_source_id: str, *, credential_id: str | None = None, unpack_archive: bool = True, overwrite: OverwriteStrategyName = 'rename') -> str

Start a data-source import and return the async status id.

Source code in datarobot_genai/drmcputils/files/file_system_store.py 
272
273
274
275
276
277
278
279
280
281
282
async def import_from_data_source(
    self,
    path: str,
    data_source_id: str,
    *,
    credential_id: str | None = None,
    unpack_archive: bool = True,
    overwrite: OverwriteStrategyName = "rename",
) -> str:
    """Start a data-source import and return the async status id."""
    ...

get_status async 

get_status(status_id: str) -> dict[str, Any]

Fetch the current async status payload for a single import job.

Source code in datarobot_genai/drmcputils/files/file_system_store.py 
284
285
286
async def get_status(self, status_id: str) -> dict[str, Any]:
    """Fetch the current async status payload for a single import job."""
    ...

DataRobotFileSystemStore

:class:FileSystemStore backed by :class:datarobot.fs.DataRobotFileSystem.

Scoped to the requesting user's DataRobot token; blocking fsspec calls run in a worker thread. A fresh filesystem is built per call inside the thread — it holds no client state (the Files SDK resolves the request-scoped client from the context configured by :func:request_user_dr_sdk).

Source code in datarobot_genai/drmcputils/files/file_system_store.py 
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
546
547
548
549
550
551
552
553
554
555
556
557
558
559
560
561
562
563
564
class DataRobotFileSystemStore:
    """:class:`FileSystemStore` backed by :class:`datarobot.fs.DataRobotFileSystem`.

    Scoped to the requesting user's DataRobot token; blocking fsspec calls run in
    a worker thread. A fresh filesystem is built per call inside the thread — it
    holds no client state (the ``Files`` SDK resolves the request-scoped client
    from the context configured by :func:`request_user_dr_sdk`).
    """

    def __init__(self, *, headers_auth_only: bool = True) -> None:
        # When False, fall back to the application API token outside HTTP contexts.
        self._headers_auth_only = headers_auth_only

    async def _run(self, fn: Callable[[Any], T]) -> T:
        """Run ``fn(fs)`` in a worker thread under the request-scoped DR client."""
        from datarobot.fs import DataRobotFileSystem

        def _call() -> T:
            return fn(DataRobotFileSystem())

        with request_user_dr_sdk(headers_auth_only=self._headers_auth_only):
            try:
                return await asyncio.to_thread(_call)
            except ClientError as exc:
                raise_tool_error_for_client_error(exc)
            except OSError as exc:
                _raise_tool_error_for_fs_error(exc)
            except ValueError as exc:
                _raise_tool_error_for_fs_error(exc)

    async def ls(self, path: str, *, detail: bool = True) -> list[FileEntry]:
        results: list[Mapping[str, Any]] = await self._run(lambda fs: fs.ls(path, detail=True))
        return [FileEntry.from_info(item) for item in results]

    async def find(
        self, path: str, *, maxdepth: int | None = None, withdirs: bool = False
    ) -> list[FileEntry]:
        results: Mapping[str, Mapping[str, Any]] = await self._run(
            lambda fs: fs.find(path, maxdepth=maxdepth, withdirs=withdirs, detail=True)
        )
        return [FileEntry.from_info(info) for info in results.values()]

    async def glob(self, pattern: str, *, maxdepth: int | None = None) -> list[FileEntry]:
        results: Mapping[str, Mapping[str, Any]] = await self._run(
            lambda fs: fs.glob(pattern, maxdepth=maxdepth, detail=True)
        )
        return [FileEntry.from_info(info) for info in results.values()]

    async def tree(
        self,
        path: str,
        *,
        recursion_limit: int = 2,
        max_display: int = 25,
        display_size: bool = False,
    ) -> str:
        return await self._run(
            lambda fs: fs.tree(
                path,
                recursion_limit=recursion_limit,
                max_display=max_display,
                display_size=display_size,
            )
        )

    async def info(self, path: str) -> FileEntry:
        info: Mapping[str, Any] = await self._run(lambda fs: fs.info(path))
        return FileEntry.from_info(info)

    async def read(self, path: str, *, start: int | None = None, end: int | None = None) -> bytes:
        return await self._run(lambda fs: fs.cat_file(path, start=start, end=end))

    async def sign(self, path: str, *, expiration: int = DEFAULT_SIGN_EXPIRATION_SECONDS) -> str:
        return await self._run(lambda fs: fs.sign(path, expiration=expiration))

    async def write(self, path: str, data: bytes, *, mode: str = "overwrite") -> None:
        await self._run(lambda fs: fs.pipe_file(path, value=data, mode=mode))

    async def upload(
        self,
        local_path: str | list[str],
        dest: str | list[str],
        *,
        recursive: bool = False,
        maxdepth: int | None = None,
        overwrite: OverwriteStrategyName = "rename",
    ) -> None:
        strategy = overwrite_strategy_from_name(overwrite)
        await self._run(
            lambda fs: fs.put(
                local_path,
                dest,
                recursive=recursive,
                maxdepth=maxdepth,
                overwrite_strategy=strategy,
            )
        )

    async def create_dir(self) -> str:
        return await self._run(lambda fs: fs.create_catalog_item_dir())

    async def delete(
        self, path: str, *, recursive: bool = False, maxdepth: int | None = None
    ) -> None:
        def _call(fs: Any) -> None:
            if not recursive:
                try:
                    info = fs.info(path)
                except FileNotFoundError:
                    info = None
                if (
                    info is not None
                    and info.get("type") == "directory"
                    and fs.ls(path, detail=False)
                ):
                    raise DirectoryNotEmptyError(
                        f"{path!r} is a non-empty directory; pass recursive=True to delete it."
                    )
            fs.rm(path, recursive=recursive, maxdepth=maxdepth)

        await self._run(_call)

    async def copy(
        self,
        source: str,
        dest: str,
        *,
        recursive: bool = False,
        maxdepth: int | None = None,
        overwrite: OverwriteStrategyName = "rename",
    ) -> None:
        strategy = overwrite_strategy_from_name(overwrite)
        await self._run(
            lambda fs: fs.copy(
                source,
                dest,
                recursive=recursive,
                maxdepth=maxdepth,
                overwrite_strategy=strategy,
            )
        )

    async def move(
        self,
        source: str,
        dest: str,
        *,
        recursive: bool = False,
        maxdepth: int | None = None,
        overwrite: OverwriteStrategyName = "rename",
    ) -> None:
        strategy = overwrite_strategy_from_name(overwrite)
        await self._run(
            lambda fs: fs.mv(
                source,
                dest,
                recursive=recursive,
                maxdepth=maxdepth,
                overwrite_strategy=strategy,
            )
        )

    async def clone(self, path_or_id: str, *, files_to_omit: list[str] | None = None) -> str:
        return await self._run(
            lambda fs: fs.clone_catalog_item_dir(path_or_id, files_to_omit=files_to_omit)
        )

    async def import_from_url(
        self,
        path: str,
        url: str,
        *,
        unpack_archive: bool = True,
        overwrite: OverwriteStrategyName = "rename",
    ) -> str:
        strategy = overwrite_strategy_from_name(overwrite)

        def _call(fs: Any) -> str:
            catalog_id, internal_path = fs._split_path(path)
            prefix = f"{internal_path.rstrip('/')}/" if internal_path else None
            entity = fs._get_files_wrapper_for_folder_id(catalog_id).upload_from_url(
                url=url,
                use_archive_contents=unpack_archive,
                overwrite=strategy,
                wait_for_completion=False,
                prefix=prefix,
            )
            location = entity._async_status_location
            if not location:
                raise ToolError(
                    "Import did not return an async status location.",
                    kind=ToolErrorKind.UPSTREAM,
                )
            return extract_status_id(location)

        return await self._run(_call)

    async def import_from_data_source(
        self,
        path: str,
        data_source_id: str,
        *,
        credential_id: str | None = None,
        unpack_archive: bool = True,
        overwrite: OverwriteStrategyName = "rename",
    ) -> str:
        strategy = overwrite_strategy_from_name(overwrite)

        def _call(fs: Any) -> str:
            catalog_id, internal_path = fs._split_path(path)
            prefix = f"{internal_path.rstrip('/')}/" if internal_path else None
            entity = fs._get_files_wrapper_for_folder_id(catalog_id).upload_from_data_source(
                data_source_id=data_source_id,
                credential_id=credential_id,
                prefix=prefix,
                use_archive_contents=unpack_archive,
                overwrite=strategy,
                wait_for_completion=False,
            )
            location = entity._async_status_location
            if not location:
                raise ToolError(
                    "Import did not return an async status location.",
                    kind=ToolErrorKind.UPSTREAM,
                )
            return extract_status_id(location)

        return await self._run(_call)

    async def get_status(self, status_id: str) -> dict[str, Any]:
        location = normalize_status_location(status_id)

        def _fetch() -> dict[str, Any]:
            client = dr.client.get_client()
            response = client.get(location, allow_redirects=False)
            if response.status_code == 307:
                redirect = response.headers.get("Location")
                if not redirect:
                    raise ToolError(
                        "Status redirect (307) missing Location header.",
                        kind=ToolErrorKind.UPSTREAM,
                    )
                response = client.get(redirect, allow_redirects=False)
            if response.status_code == 303:
                return {
                    "status": FILE_IMPORT_COMPLETED_STATUS,
                    "redirect_location": response.headers.get("Location"),
                }
            return response.json()

        with request_user_dr_sdk(headers_auth_only=self._headers_auth_only):
            try:
                return await asyncio.to_thread(_fetch)
            except ClientError as exc:
                raise_tool_error_for_client_error(exc)

normalize_status_location 

normalize_status_location(status_id: str) -> str

Return a relative API route for status/<id>/ from a bare id, route, or URL.

Source code in datarobot_genai/drmcputils/files/file_system_store.py 
83
84
85
86
87
88
89
90
91
92
def normalize_status_location(status_id: str) -> str:
    """Return a relative API route for ``status/<id>/`` from a bare id, route, or URL."""
    cleaned = status_id.strip()
    if not cleaned:
        raise ToolError(
            "Argument validation error: 'status_id' cannot be empty.",
            kind=ToolErrorKind.VALIDATION,
        )
    bare_id = extract_status_id(cleaned)
    return f"status/{bare_id}/"

extract_status_id 

extract_status_id(async_location: str) -> str

Extract the bare status id from an async Location header or route.

Source code in datarobot_genai/drmcputils/files/file_system_store.py 
 95
 96
 97
 98
 99
100
101
102
103
def extract_status_id(async_location: str) -> str:
    """Extract the bare status id from an async Location header or route."""
    location = async_location.strip().rstrip("/")
    if location.startswith("http://") or location.startswith("https://"):
        path = urlparse(location).path.rstrip("/")
        location = path.split("/api/v2/", 1)[-1] if "/api/v2/" in path else path.lstrip("/")
    if location.startswith("status/"):
        return location.split("/", 1)[1].split("/", 1)[0]
    return location.split("/")[-1]