Remote Storage

┌─────────────────┐
│  Your Workspace │  (working files)
│   data/         │
└────────┬────────┘
         │ dvc add/checkout
         ↓
┌─────────────────┐
│  Local Cache    │  (.dvc/cache)
│  Content-based  │
│  storage        │
└────────┬────────┘
         │ dvc push/pull
         ↓
┌─────────────────┐
│ Remote Storage  │  (S3, GCS, Azure, etc.)
│  Team's shared  │
│  cache          │
└─────────────────┘

# Amazon S3
dvc remote add -d myremote s3://mybucket/dvc-storage

# Google Cloud Storage
dvc remote add -d myremote gs://mybucket/dvc-storage

# Azure Blob Storage
dvc remote add -d myremote azure://mycontainer/path

# SSH
dvc remote add -d myremote ssh://[email protected]/path/to/storage

# Local or network drive
dvc remote add -d myremote /mnt/shared-storage

class DataCloud:
    """Class that manages dvc remotes.
    
    Args:
        repo (dvc.repo.Repo): repo instance that belongs to the repo that
            we are working on.
    
    Raises:
        config.ConfigError: thrown when config has invalid format.
    """
    
    def __init__(self, repo):
        self.repo = repo
    
    def get_remote(
        self,
        name: Optional[str] = None,
        command: str = "<command>",
    ) -> "Remote":
        if not name:
            name = self.repo.config["core"].get("remote")
        
        if name:
            from dvc.fs import get_cloud_fs
            
            cls, config, fs_path = get_cloud_fs(self.repo.config, name=name)
            # ... create and return Remote instance

class Remote:
    def __init__(self, name: str, path: str, fs: "FileSystem", *, index=None, **config):
        self.path = path
        self.fs = fs
        self.name = name
        self.index = index
        
        self.worktree: bool = config.pop("worktree", False)
        self.config = config
    
    @cached_property
    def odb(self) -> "HashFileDB":
        from dvc.cachemgr import CacheManager
        from dvc_data.hashfile.db import get_odb
        from dvc_data.hashfile.hash import DEFAULT_ALGORITHM
        
        path = self.path
        if self.worktree:
            path = self.fs.join(path, ".dvc", CacheManager.FILES_DIR, DEFAULT_ALGORITHM)
        else:
            path = self.fs.join(path, CacheManager.FILES_DIR, DEFAULT_ALGORITHM)
        return get_odb(self.fs, path, hash_name=DEFAULT_ALGORITHM, **self.config)

# Push all tracked data
dvc push

# Push specific files
dvc push data/train.csv.dvc

# Push to specific remote
dvc push -r myremote

# Push with multiple parallel jobs
dvc push -j 8

def push(
    self,
    objs: Iterable["HashInfo"],
    jobs: Optional[int] = None,
    remote: Optional[str] = None,
    odb: Optional["HashFileDB"] = None,
) -> "TransferResult":
    """Push data items in a cloud-agnostic way.
    
    Args:
        objs: objects to push to the cloud.
        jobs: number of jobs that can be running simultaneously.
        remote: optional name of remote to push to.
            By default remote from core.remote config option is used.
        odb: optional ODB to push to. Overrides remote.
    """
    if odb is not None:
        return self._push(objs, jobs=jobs, odb=odb)
    legacy_objs, default_objs = _split_legacy_hash_infos(objs)
    result = TransferResult(set(), set())
    if legacy_objs:
        odb = self.get_remote_odb(remote, "push", hash_name="md5-dos2unix")
        t, f = self._push(legacy_objs, jobs=jobs, odb=odb)
        result.transferred.update(t)
        result.failed.update(f)
    if default_objs:
        odb = self.get_remote_odb(remote, "push")
        t, f = self._push(default_objs, jobs=jobs, odb=odb)
        result.transferred.update(t)
        result.failed.update(f)
    return result

# Pull all tracked data
dvc pull

# Pull specific files
dvc pull data/train.csv.dvc

# Pull from specific remote
dvc pull -r myremote

# Pull with multiple parallel jobs
dvc pull -j 8

# Check status against default remote
dvc status -c

# Check against specific remote
dvc status -r myremote -c

[core]
    remote = myremote

['remote "myremote"']
    url = s3://mybucket/dvc-storage
    region = us-west-2
    profile = myprofile

# Set remote-specific options
dvc remote modify myremote region us-west-2
dvc remote modify myremote profile myprofile

# For S3
dvc remote modify myremote access_key_id YOUR_KEY
dvc remote modify myremote secret_access_key YOUR_SECRET

# Use AWS credentials file
dvc remote modify myremote profile myprofile

# Use environment variables
export AWS_ACCESS_KEY_ID="..."
export AWS_SECRET_ACCESS_KEY="..."

# Use IAM role (on EC2)
# No configuration needed

# Use service account
export GOOGLE_APPLICATION_CREDENTIALS="path/to/credentials.json"

# Or configure explicitly
dvc remote modify myremote credentialpath path/to/credentials.json

# Use connection string
dvc remote modify myremote connection_string "..."

# Or use account name and key
dvc remote modify myremote account_name myaccount
dvc remote modify myremote account_key "..."

# Use SSH key
dvc remote modify myremote keyfile ~/.ssh/id_rsa

# Use password (not recommended)
dvc remote modify myremote password mypassword

# Use SSH agent
dvc remote modify myremote ask_password true

dvc remote modify myremote version_aware true

dvc remote add -d shared /mnt/shared-dvc-repo
dvc remote modify shared worktree true

dvc remote modify myremote read_only true

# Store by branch
dvc remote modify myremote url s3://bucket/${DVC_EXP_NAME}

# Store by user
dvc remote modify myremote url s3://bucket/${USER}

# Use more parallel transfers
dvc push -j 16
dvc pull -j 16

# Pull specific pipeline stage
dvc repro --pull train

# Pull only metrics (small files)
dvc pull --run-cache

# Increase retry attempts for unreliable connections
dvc remote modify myremote retry_count 10

# Increase timeout
dvc remote modify myremote timeout 300

def _split_legacy_hash_infos(
    hash_infos: Iterable["HashInfo"],
) -> tuple[set["HashInfo"], set["HashInfo"]]:
    from dvc.cachemgr import LEGACY_HASH_NAMES
    
    legacy = set()
    default = set()
    for hi in hash_infos:
        if hi.name in LEGACY_HASH_NAMES:
            legacy.add(hi)
        else:
            default.add(hi)
    return legacy, default

def transfer(
    self,
    src_odb: "HashFileDB",
    dest_odb: "HashFileDB",
    objs: Iterable["HashInfo"],
    **kwargs,
) -> "TransferResult":
    from dvc_data.hashfile.transfer import transfer
    
    return transfer(src_odb, dest_odb, objs, **kwargs)

# Default remote for team
dvc remote add -d team s3://team-bucket/dvc-storage

# Personal backup
dvc remote add backup gs://my-personal-bucket/backup

# Local cache for quick access
dvc remote add local /mnt/fast-storage

# Push to all remotes
dvc push -r team
dvc push -r backup

# Test remote connectivity
dvc remote list
dvc status -c -r myremote

# Increase verbosity
dvc push -v
dvc pull -vv

# Verify credentials
aws s3 ls s3://mybucket/  # For S3
gsutil ls gs://mybucket/  # For GCS

# Check DVC configuration
dvc config remote.myremote.url

# Use more parallel jobs
dvc push -j 32

# Skip checksum verification (faster but risky)
dvc push --no-verify