ファイルレジストリ¶

pip install 'datarobot[fs]' 

import datarobot as dr
from datarobot.fs import DataRobotFileSystem

dr.Client(token="<YOUR_API_TOKEN>", endpoint="https://app.datarobot.com/api/v2")

fs = DataRobotFileSystem() 

# Create a brand-new, empty catalog item
catalog_id = fs.create_catalog_item_dir() 

# Clone an existing catalog item, copying every file into a new one
source_catalog_id = "<EXISTING_CATALOG_ITEM_ID>"
clone_id = fs.clone_catalog_item_dir(source_catalog_id)

# Or clone but omit specific files from the source
partial_clone_id = fs.clone_catalog_item_dir(
    source_catalog_id,
    files_to_omit=["data/scores.csv", "notes/draft.txt"],
) 

# Write a text file in place
with fs.open(f"dr://{catalog_id}/notes/readme.txt", mode="w") as f:
    f.write("This catalog item contains demo files for the file system guide.")

# Write raw bytes in a single call.
fs.pipe_file(f"dr://{catalog_id}/data/sample.csv", b"name,score\nCharlie,72\n") 

from datarobot.enums import FilesOverwriteStrategy

# Write to the existing path notes/readme.txt. 
# The new content will be placed in a new file /notes/readme (2).txt
with fs.open(
    f"dr://{catalog_id}/notes/readme.txt",
    mode="w",
    overwrite_strategy=FilesOverwriteStrategy.RENAME
) as f:
    f.write("This content is written to a new file because RENAME was specified.") 

import tempfile
import fsspec

# Use the local fsspec implementation to stage demo files in a temp directory.
local_fs = fsspec.filesystem("local")
local_dir = tempfile.mkdtemp()

local_fs.makedirs(f"{local_dir}/notes", exist_ok=True)
with local_fs.open(f"{local_dir}/scores.csv", "w") as f:
    f.write("name,score\nAlice,95\nBob,87\n")
with local_fs.open(f"{local_dir}/notes/agenda.txt", "w") as f:
    f.write("Q3 planning agenda")
with local_fs.open(f"{local_dir}/notes/actions.txt", "w") as f:
    f.write("1. Review roadmap\n2. Confirm budget\n")

# Upload a single file
fs.put_file(f"{local_dir}/scores.csv", f"dr://{catalog_id}/data/scores.csv")

# Upload a directory recursively. Trailing slashes mark both paths as directories.
fs.put(f"{local_dir}/notes/", f"dr://{catalog_id}/notes/", recursive=True) 

# Ingest from url and create file dr://<catalog-id>/external/iris.csv
fs.put_from_url(
    path=f"dr://{catalog_id}/external/",
    url="https://s3.amazonaws.com/datarobot_public_datasets/iris.csv",
) 

credential = dr.Credential.create_s3(
    name="S3 Credential",
    aws_access_key_id="<AWS_ACCESS_KEY_ID>",
    aws_secret_access_key="<AWS_SECRET_ACCESS_KEY>",
)
s3_connector = next(c for c in dr.Connector.list() if c.connector_type == "s3")

s3_data_store = dr.DataStore.create(
    data_store_type=dr.enums.DataStoreTypes.DR_CONNECTOR_V1,
    canonical_name="My S3 Bucket",
    fields=[
        {"id": "fs.defaultFS", "name": "Bucket Name", "value": "my-bucket-name"},
        {"id": "fs.rootDirectory", "name": "Prefix", "value": "/"},
        {"id": "fs.s3.awsRegion", "name": "S3 Bucket Region", "value": "us-east-1"},
    ],
    connector_id=s3_connector.id,
)
s3_data_source = dr.DataSource.create(
    data_source_type=dr.enums.DataStoreTypes.DR_CONNECTOR_V1,
    canonical_name="S3 Documents",
    params=dr.DataSourceParameters(
        data_store_id=s3_data_store.id,
        path="documents/",
    ),
)

fs.put_from_data_source(
    path=f"dr://{catalog_id}/s3_documents/",
    data_source_id=s3_data_source.id,
    credential_id=credential.credential_id,
) 

# List the immediate contents of the catalog item. Set detail=False for just the paths.
fs.ls(f"dr://{catalog_id}/", detail=False)

# Use detail=True (the default) to also retrieve size, type, and format.
for item in fs.ls(f"dr://{catalog_id}/", detail=True):
    print(f"{item['name']:50s} type={item['type']:10s} size={item['size']}")

# Recursively list every file. Pass withdirs=True to include directories.
all_files = fs.find(f"dr://{catalog_id}/")

# Walk the directory tree one level at a time, similar to os.walk().
for dirpath, dirnames, filenames in fs.walk(f"dr://{catalog_id}/"):
    print((dirpath, dirnames, filenames))

# Glob lets you match by pattern. Supports *, **, ?, and [abc] character classes.
csv_files = fs.glob(f"dr://{catalog_id}/**/*.csv")

# Visualize the catalog item layout. The recursion_limit controls how deep to walk.
print(fs.tree(f"dr://{catalog_id}/", recursion_limit=3)) 

from datarobot.enums import FilesOverwriteStrategy

# Copy a single file.
fs.copy(
    f"dr://{catalog_id}/data/scores.csv",
    f"dr://{catalog_id}/backups/scores_backup.csv",
)

# Copy a directory recursively, skipping any files that already exist in the destination.
# Both paths end with / to mark them as directories.
fs.copy(
    f"dr://{catalog_id}/notes/",
    f"dr://{catalog_id}/archive/notes_snapshot/",
    recursive=True,
    overwrite_strategy=FilesOverwriteStrategy.SKIP
)

# Use a glob pattern to copy all .txt files into a single folder.
fs.copy(
    f"dr://{catalog_id}/**/*.txt",
    f"dr://{catalog_id}/all_text_files/",
    recursive=True,
) 

# Rename a file by moving it to a new path within the same catalog item.
fs.mv(f"dr://{catalog_id}/backups/scores_backup.csv", f"dr://{catalog_id}/backups/scores_v1.csv")

# Move a file into a different directory (note the trailing slash on the target).
fs.mv(f"dr://{catalog_id}/backups/scores_v1.csv", f"dr://{catalog_id}/archive/") 

# Delete a single file.
fs.rm(f"dr://{catalog_id}/archive/scores_v1.csv")

# Delete a directory recursively.
fs.rm(f"dr://{catalog_id}/all_text_files/", recursive=True)

# Delete every csv file under archive.
fs.rm(f"dr://{catalog_id}/archive/**/*.csv", recursive=True) 

# Iterate line by line. Never loads the full file into memory.
with fs.open(f"dr://{catalog_id}/data/scores.csv", mode="r") as f:
    for line in f:
        print(line.rstrip())

# Read in binary mode with seeking.
with fs.open(f"dr://{catalog_id}/data/scores.csv", mode="rb") as f:
    header = f.read(20)
    f.seek(0)
    full = f.read() 

# Read a single file as bytes.
data = fs.cat(f"dr://{catalog_id}/data/scores.csv")

# Read every CSV file in the catalog item at once.
all_csvs = fs.cat(f"dr://{catalog_id}/**/*.csv", recursive=True)
for path, content in all_csvs.items():
    print(f"{path}: {len(content)} bytes") 

import requests

url = fs.sign(f"dr://{catalog_id}/data/scores.csv", expiration=300)

# Download file locally using signed url
local_path = "scores.csv"
with requests.get(url, stream=True) as r:
    r.raise_for_status()
    with open(local_path, "wb") as f:
        for chunk in r.iter_content(chunk_size=8192):
            if chunk:
                f.write(chunk) 

import tempfile

local_dir = tempfile.mkdtemp()

# Download a single file to a local path.
fs.get_file(f"dr://{catalog_id}/data/scores.csv", f"{local_dir}/scores.csv")

# Download a directory recursively. Trailing slashes mark both paths as directories.
fs.get(f"dr://{catalog_id}/notes/", f"{local_dir}/notes/", recursive=True)

# Use a glob pattern to download all .csv files into a single local directory.
fs.get(f"dr://{catalog_id}/**/*.csv", f"{local_dir}/all_csvs/", recursive=True) 

# Detailed metadata for a single file or directory.
file_info = fs.info(f"dr://{catalog_id}/data/scores.csv")

# Quick existence checks.
print("exists?", fs.exists(f"dr://{catalog_id}/data/scores.csv"))
print("isfile?", fs.isfile(f"dr://{catalog_id}/data/scores.csv"))
print("isdir?",  fs.isdir(f"dr://{catalog_id}/data/"))

# Total disk usage for the entire catalog item.
print(f"Total bytes: {fs.du(f'dr://{catalog_id}/', total=True):,}") 

mapper = fs.get_mapper(f"dr://{catalog_id}/data/")

# List files in mapping
print("Keys:", list(mapper))
# Read bytes from a file
print("scores.csv first 30 bytes:", mapper["scores.csv"][:30])

# Write to create a new file
mapper["generated.txt"] = b"This file was created via the mapper interface."

# Membership checks and size.
print("'scores.csv' in mapper?", "scores.csv" in mapper)
print("Total files:", len(mapper))

# Delete file
del mapper["generated.txt"]