How to Use#
Install the package#
pip install core-ftp
uv pip install core-ftp # Or using UV...
Examples…#
Password authentication
from core_ftp.clients.sftp import SftpClient, SftpConnectionConfig
config = SftpConnectionConfig(host="test.rebex.net", port=22, user="demo", password="password")
with SftpClient(config) as client:
for name, attr in client.list_files("/"):
print(name, attr)
SSH key authentication
from core_ftp.clients.sftp import SftpClient, SftpConnectionConfig
config = SftpConnectionConfig(host="localhost", port=23, user="foo", private_key_path="/path/to/id_rsa")
with SftpClient(config) as client:
client.download_file("remote/file.txt", "/local/file.txt")
Upload a file
from core_ftp.clients.sftp import SftpClient, SftpConnectionConfig
config = SftpConnectionConfig(host="localhost", user="foo", password="pass")
with SftpClient(config) as client:
client.chdir("/upload")
client.upload_file("/local/report.csv", "report.csv", confirm=True)
ETL task (file-based processing)
from core_ftp.clients.sftp import SftpConnectionConfig
from core_ftp.etls.file_based import IBaseEtlFromFtpFile, SftpFileConfig
class DataProcessor(IBaseEtlFromFtpFile):
@classmethod
def registered_name(cls) -> str:
return "DataProcessor"
def process_file(self, path: str, **kwargs):
print(f"Processing {path}")
DataProcessor(
connection=SftpConnectionConfig(host="data.server.com", user="admin", password="secret"),
file_config=SftpFileConfig(path="/data/csv", file_ext=".csv", delete_file_on_success=True),
).execute()
Local SFTP server using Docker#
Functional tests require two local SFTP servers running via Docker (atmoz/sftp): one for password authentication (port 22) and one for SSH key authentication (port 23).
Note
Docker Desktop users (macOS / Windows / Linux): Docker Desktop runs inside a VM and can only bind-mount paths that are explicitly shared with it. Before starting the containers, add the following paths in Docker Desktop → Settings → Resources → File Sharing, then click Apply & Restart:
<project_root>/tests/resources/upload<project_root>/tests/resources/ssh_keys
Without this step the containers will fail with “bind source path does not exist” or “path is not shared from the host”.
1. Password authentication (port 22)
docker run \
--mount type=bind,source="$(pwd)/tests/resources/upload",target=/home/foo/upload \
-p 22:22 -d atmoz/sftp foo:pass:::upload
2. SSH key authentication (port 23)
docker run \
--mount type=bind,source="$(pwd)/tests/resources/ssh_keys/id_rsa.pub",target=/home/foo/.ssh/keys/id_rsa.pub,readonly \
--mount type=bind,source="$(pwd)/tests/resources/upload",target=/home/foo/upload \
-p 23:22 -d atmoz/sftp foo::1001
Run functional tests
python manager.py run-tests --test-type functional --pattern "*.py"