FTPC Storage (Full Access)

Use the

ftpc

This skill can modify remote storage. Confirm destructive operations with the user.

Supported Backends

file:///path

/path

/home/user/data

ftp://[user:pass@]host[:port]/path

ftp://ftp.example.com/pub

ftps://[user:pass@]host[:port]/path

ftps://secure.example.com

sftp://[user:pass@]host[:port]/path

sftp://user:pass@host/data

s3://bucket/path

s3://my-bucket/folder

azure://account.dfs.core.windows.net/fs/path

azure://myacct.dfs.core.windows.net/data

blob://account.blob.core.windows.net/container/path

blob://myacct.blob.core.windows.net/files

Quick Start

from ftpc import connect_sync

with connect_sync("s3://my-bucket") as store:
    # List files
    files = store.list("/")

    # Download
    store.download("/data.csv", "local.csv")

    # Upload
    store.upload("report.pdf", "/reports/report.pdf")

    # Create directory
    store.mkdir("/new-folder")

    # Delete file
    store.delete("/old-file.txt")

All Operations

List Directory

from ftpc import connect_sync

with connect_sync("ftp://ftp.example.com") as store:
    files = store.list("/documents")
    for f in files:
        print(f"{f.name}  {'DIR' if f.is_directory else f.size}")

Download File

with connect_sync("sftp://user:pass@host") as store:
    store.download("/remote/file.csv", "local_file.csv")

    # With progress callback
    def progress(bytes_done: int) -> bool:
        print(f"Downloaded {bytes_done} bytes")
        return True  # False cancels transfer

    store.download("/large.zip", "out.zip", progress)

Upload File

with connect_sync("s3://my-bucket") as store:
    store.upload("local_report.pdf", "/reports/2024/report.pdf")

    # With progress
    def progress(bytes_done: int) -> bool:
        print(f"Uploaded {bytes_done} bytes")
        return True

    store.upload("large_backup.tar.gz", "/backups/backup.tar.gz", progress)

Create Directory

with connect_sync("ftp://ftp.example.com") as store:
    success = store.mkdir("/new-directory")
    if success:
        print("Directory created")

Delete File

with connect_sync("s3://my-bucket") as store:
    success = store.delete("/obsolete-file.txt")
    if success:
        print("File deleted")

Async Usage

import asyncio
from ftpc import Storage

async def main():
    async with Storage.connect("s3://bucket") as store:
        await store.upload("local.txt", "/remote.txt")
        await store.mkdir("/new-dir")
        await store.delete("/old.txt")

asyncio.run(main())

Named Constructors

from ftpc import Storage

# S3 with credentials
with Storage.s3(
    bucket="my-bucket",
    region="us-east-1",
    access_key_id="AKIAIOSFODNN7EXAMPLE",
    secret_access_key="wJalrXUtnFEMI/K7MDENG/bPxRfiCYEXAMPLEKEY"
).sync() as store:
    store.upload("data.csv", "/uploads/data.csv")

# FTP with TLS
with Storage.ftp(
    host="ftp.example.com",
    username="user",
    password="pass",
    tls=True
).sync() as store:
    store.upload("report.pdf", "/reports/report.pdf")

# SFTP with key auth
with Storage.sftp(
    host="server.example.com",
    username="deploy",
    key_filename="/home/user/.ssh/id_rsa"
).sync() as store:
    store.upload("deploy.tar.gz", "/var/releases/deploy.tar.gz")

# Azure Data Lake
with Storage.azure(
    account_url="https://myaccount.dfs.core.windows.net",
    filesystem="data",
    account_key="your-account-key"
).sync() as store:
    store.upload("dataset.parquet", "/datasets/dataset.parquet")

Using Named Remotes from Config

With

~/.ftpcconf.toml

[prod-s3]
type = "s3"
bucket = "production-data"
region = "us-east-1"

from ftpc.config import Config
from ftpc.clients.s3client import S3Client

config = Config.from_file()
remote = config.remotes["prod-s3"]

with S3Client(bucket_name=remote.bucket, region_name=remote.region) as client:
    client.put("local.csv", "/uploads/data.csv")

FileDescriptor Structure

All

list()

List[FileDescriptor]

@dataclass
class FileDescriptor:
    path: PurePath              # Full path
    filetype: FileType          # FILE or DIRECTORY
    size: Optional[int]         # Bytes (None for dirs)
    modified_time: Optional[datetime]

    @property
    def name(self) -> str       # Filename only

    @property
    def is_file(self) -> bool

    @property
    def is_directory(self) -> bool

Dependencies

pip install ftpc            # Core (local + FTP)
pip install ftpc[sftp]      # + SFTP
pip install ftpc[s3]        # + S3
pip install ftpc[azure]     # + Azure
pip install ftpc[all]       # All backends