Request Parameters
- username
- password
- token (optional, supplied instead of password)
- env (optional, one of
""(default, production),"int","dev") - target (optional,
"drive"(default) or"bucket"— selects which backend to talk to)
Sample Case
import ebrains_drive
# Drive (Seafile) — default target
client = ebrains_drive.connect('hbp_username', 'password')
# Or talk to the Bucket (Data Proxy) backend via the same factory
bucket_client = ebrains_drive.connect('hbp_username', 'password', target="bucket")Return Type
A DriveApiClient when target="drive" (default), a BucketApiClient
when target="bucket". For anonymous read access to public buckets,
construct BucketApiClient() directly — connect() always authenticates.
Request Parameters
- repo_id
Sample Case
import ebrains_drive
client = ebrains_drive.connect('hbp_username', 'password')
repo = client.repos.get_repo('09c16e2a-ff1a-4207-99f3-1351c3f1e507')
# `client.repos.get(repo_id)` is an alias that mirrors the Bucket surface.Return Type
A Library Object
Exception
- Library does not exist.
Request Parameters
None
Sample Case
import ebrains_drive
client = ebrains_drive.connect('hbp_username', 'password')
repo = client.repos.get_repo('09c16e2a-ff1a-4207-99f3-1351c3f1e507')
is_readonly = repo.is_readonly()Bucket.is_readonly() exists with the same signature on the Bucket
backend, so cross-backend code can call container.is_readonly()
regardless of which container type it has.
Return Type
Boolean
Request Parameters
None
Sample Case
import ebrains_drive
client = ebrains_drive.connect('hbp_username', 'password')
repo_list = client.repos.list_repos()
print(repo_list)
Out >>> [<ebrains_drive.repo.Repo at 0x7f1bb0769750>,
<ebrains_drive.repo.Repo at 0x7f1bb07693d0>,
<ebrains_drive.repo.Repo at 0x7f1bb0769a50>,
<ebrains_drive.repo.Repo at 0x7f1bb077cc10>,
<ebrains_drive.repo.Repo at 0x7f1bb077cfd0>,
<ebrains_drive.repo.Repo at 0x7f1bb077ca10>]
print([repo.name for repo in repo_list])
Out >>> ['alphabox',
'hello',
'Doc',
'obj_test',
'fs_test',
'global']
# `client.repos.list()` is an alias for `list_repos()`.Return Type
A list of Libraries Object
Request Parameters
- name
- password (default None)
Sample Case
import ebrains_drive
client = ebrains_drive.connect('hbp_username', 'password')
repo = client.repos.create_repo('test_repo')
# `client.repos.create('test_repo')` is an alias.Return Type
A Library Object
Request Parameters
None
Sample Case
import ebrains_drive
client = ebrains_drive.connect('hbp_username', 'password')
repo = client.repos.get_repo('09c16e2a-ff1a-4207-99f3-1351c3f1e507')
repo.delete()
# Equivalently: client.repos.delete('09c16e2a-...').Return Type
None
Request Parameters
- filelike_or_path (file-like object,
bytes, or path to a local file) - filename (name to store the file under, at the root of the library)
Sample Case
import ebrains_drive
client = ebrains_drive.connect('hbp_username', 'password')
repo = client.repos.get_repo('09c16e2a-ff1a-4207-99f3-1351c3f1e507')
# Path or file-like — both work
repo.upload('/home/jovyan/notes.md', 'notes.md')
repo.upload_local_file('/home/jovyan/notes.md', overwrite=True)These are convenience shortcuts for repo.get_dir("/").upload(...) and
repo.get_dir("/").upload_local_file(...) that mirror
Bucket.upload / Bucket.upload_local_file.
Return Type
A SeafFile of the newly uploaded file.
Request Parameters
- path
Sample Case
import ebrains_drive
client = ebrains_drive.connect('hbp_username', 'password')
repo = client.repos.get_repo('09c16e2a-ff1a-4207-99f3-1351c3f1e507')
seafdir = repo.get_dir('/root')
print(seafdir.__dict__)
Out >>> {'client': DriveApiClient[server=http://127.0.0.1:8000, user=admin@admin.com],
'entries': [],
'id': 'c3742dd86004d51c358845fa3178c87e4ab3aa60',
'path': '/root',
'repo': <ebrains_drive.repo.Repo at 0x7f2af56b1490>,
'size': 0}Return Type
A Directory Object
Exception
- Directory does not exist.
Request Parameters
- force_refresh (default False)
Sample Case
import ebrains_drive
client = ebrains_drive.connect('hbp_username', 'password')
repo = client.repos.get_repo('09c16e2a-ff1a-4207-99f3-1351c3f1e507')
seafdir = repo.get_dir('/root')
lst = seafdir.ls(force_refresh=True)
print(lst)
Out >>> [SeafDir[repo=01ccc4,path=/Seahub/6.1.x,entries=14],
SeafDir[repo=01ccc4,path=/Seahub/6.2.2-pro,entries=1],
SeafDir[repo=01ccc4,path=/Seahub/6.2.3,entries=15],
SeafDir[repo=01ccc4,path=/Seahub/6.2.x,entries=5],
SeafFile[repo=01ccc4,path=/Seahub/.DS_Store,size=6148],
SeafFile[repo=01ccc4,path=/Seahub/error.md,size=127],
SeafFile[repo=01ccc4,path=/Seahub/preview-research.md,size=1030]]
print([dirent.name for dirent in lst])
Out >>> ['6.1.x',
'6.2.2-pro',
'6.2.3',
'6.2.x',
'.DS_Store',
'error.md',
'preview-research.md']Return Type
List of Directory and File
Request Parameters
- name
Sample Case
import ebrains_drive
client = ebrains_drive.connect('hbp_username', 'password')
repo = client.repos.get_repo('09c16e2a-ff1a-4207-99f3-1351c3f1e507')
seafdir = repo.get_dir('/root')
new_dir = seafdir.mkdir('tmp_dir')Return Type
A Directory Object of new directory
Request Parameters
- name (optional)
Sample Case
import ebrains_drive
client = ebrains_drive.connect('hbp_username', 'password')
repo = client.repos.get_repo('09c16e2a-ff1a-4207-99f3-1351c3f1e507')
# download the entire repository
base_dir = repo.get_dir('/')
base_dir.download()
# download a directory
my_dir = repo.get_dir("/d1/d2/dir2_1")
my_dir.download(name="somename.zip")Return Type
A Directory Object
Request Parameters
None
Sample Case
import ebrains_drive
client = ebrains_drive.connect('hbp_username', 'password')
repo = client.repos.get_repo('09c16e2a-ff1a-4207-99f3-1351c3f1e507')
seafdir = repo.get_dir('/root')
seafdir.delete()Return Type
A Response Instance
Request Parameters
- path
Sample Case
import ebrains_drive
client = ebrains_drive.connect('hbp_username', 'password')
repo = client.repos.get_repo('09c16e2a-ff1a-4207-99f3-1351c3f1e507')
seaffile = repo.get_file('/root/test.md')
print(seafile.__dict__)
Out >>> {'client': DriveApiClient[server=http://127.0.0.1:8000, user=admin@admin.com],
'id': '0000000000000000000000000000000000000000',
'path': '/root/test.md',
'repo': <ebrains_drive.repo.Repo at 0x7f2af56b1490>,
'size': 0}Return Type
A File Object
Exception
- File does not exist.
Request Parameters
- progress (keyword-only, default
False) — whenTrue, stream the download and display atqdmprogress bar.
Sample Case
import ebrains_drive
client = ebrains_drive.connect('hbp_username', 'password')
repo = client.repos.get_repo('09c16e2a-ff1a-4207-99f3-1351c3f1e507')
seaffile = repo.get_file('/root/test.md')
content = seaffile.get_content()
content = seaffile.get_content(progress=True) # with progress barThe progress= keyword mirrors DataproxyFile.get_content(progress=...)
on the Bucket backend.
Return Type
bytes (file content)
Request Parameters
- name
Sample Case
import ebrains_drive
client = ebrains_drive.connect('hbp_username', 'password')
repo = client.repos.get_repo('09c16e2a-ff1a-4207-99f3-1351c3f1e507')
seafdir = repo.get_dir('/root')
new_file = seafdir.create_empty_file('tmp_file.md')Return Type
A File Object of new empty file
Request Parameters
upload_local_file:
- filepath (path to the local file)
- name (default
None, defaults to the basename offilepath) - overwrite (default
False; whenTrue, replace an existing file with the same name)
upload (polymorphic):
- fileobj — a file-like object, a
bytesvalue, or a path/PathLike(the path overload mirrorsBucket.upload) - filename — the name to store the file under
Sample Case
import ebrains_drive
client = ebrains_drive.connect('hbp_username', 'password')
repo = client.repos.get_repo('09c16e2a-ff1a-4207-99f3-1351c3f1e507')
seafdir = repo.get_dir('/root')
# original two-method form
file = seafdir.upload_local_file('/home/ubuntu/env.md')
# `upload(...)` now also accepts a path directly
file = seafdir.upload('/home/ubuntu/env.md', 'env.md')
# ...and still accepts a file-like object / bytes
with open('/home/ubuntu/env.md', 'rb') as fp:
file = seafdir.upload(fp, 'env.md')Return Type
A File Object of the uploaded file
Exception
- Local file does not exist.
FileExistsError— whenupload_local_file(..., overwrite=False)is called with a name that already exists in the directory.
Request Parameters
- dst_dir (destination directory path)
- dst_repo_id (optional; copy/move into another library)
Sample Case
import ebrains_drive
client = ebrains_drive.connect('hbp_username', 'password')
repo = client.repos.get_repo('09c16e2a-ff1a-4207-99f3-1351c3f1e507')
seaffile = repo.get_file('/root/test.md')
# camelCase (legacy) and snake_case (new) are interchangeable
seaffile.copyTo('/backup')
seaffile.copy_to('/backup')
seaffile.move_to('/archive')Return Type
bool — True on success.
Request Parameters
None
Sample Case
import ebrains_drive
client = ebrains_drive.connect('hbp_username', 'password')
repo = client.repos.get_repo('09c16e2a-ff1a-4207-99f3-1351c3f1e507')
seaffile = repo.get_file('/root/test.md')
seaffile.delete()Return Type
A Response Instance
Request Parameters
- username (optional)
- password (optional)
- token (optional)
- env (optional)
Sample Case
# Direct construction
from ebrains_drive import BucketApiClient
client = BucketApiClient(token="ey...")
# Or via the shared factory with target="bucket"
import ebrains_drive
client = ebrains_drive.connect(token="ey...", target="bucket")
# Anonymous read access to public buckets (no token)
anon = BucketApiClient()Return Type
A BucketApiClient. client.buckets is the bucket manager (mirrors
client.repos on the Drive side); client.file exposes a URL-resolution
helper (see URL Helper).
Request Parameters
- existing_collab_name
- public (keyword-only, default
False)
Sample Case
from ebrains_drive import BucketApiClient
client = BucketApiClient(token="ey...")
bucket = client.buckets.get_bucket("existing_collab_name")
# `client.buckets.get("existing_collab_name")` is an alias.Return Type
A Bucket Object
Exceptions
- Bucket does not exist or not authorized to use the specified bucket
Request Parameters
- search (optional substring filter on bucket name)
Sample Case
from ebrains_drive import BucketApiClient
client = BucketApiClient(token="ey...")
# all buckets the credentials can access
buckets = client.buckets.list_buckets()
print([b.name for b in buckets])
# filter by name substring
atlas_buckets = client.buckets.list_buckets(search="atlas")
# `client.buckets.list()` is an alias.The returned Bucket objects have objects_count and bytes set to
None. Call client.buckets.get_bucket(name) (or .stat) to fetch the
full statistics for a specific bucket.
Return Type
A list of Bucket Objects (each carrying name, role, is_public).
Exceptions
Unauthorizedwhen the credentials cannot enumerate buckets.
Request Parameters
- bucket_name
- title (optional, defaults to
bucket_name) - description (optional)
Sample Case
from ebrains_drive import BucketApiClient
client = BucketApiClient(token="ey...")
# Canonical entry point
bucket = client.buckets.create_bucket("new_collab_name")
# `client.buckets.create("new_collab_name")` is an alias.
# Legacy form (still works, but emits a DeprecationWarning):
bucket = client.create_new("new_collab_name")Return Type
A Bucket Object
Exceptions
- Unauthorized to create new collab or bucket
Request Parameters
- bucket_name
- delete_wiki (keyword-only, default
False; whenTrue, also delete the associated wiki/collab)
Sample Case
from ebrains_drive import BucketApiClient
client = BucketApiClient(token="ey...")
# Three equivalent ways to delete a bucket:
client.buckets.delete_bucket("name") # canonical, manager-level
client.buckets.delete("name", delete_wiki=True) # alias of delete_bucket
client.buckets.get_bucket("name").delete() # instance method on Bucket
# Legacy `client.delete_bucket("name")` still works but emits a
# DeprecationWarning.Return Type
None
Exceptions
Unauthorizedwhen the credentials cannot delete the bucket.
Request Parameters
None
Sample Case
bucket = client.buckets.get_bucket("existing_collab_name")
if bucket.is_readonly():
...Returns True when bucket.role is None or "viewer"; False for
"editor" and "administrator". Dataset buckets and anonymously
accessed public buckets conservatively report read-only. Mirrors
Repo.is_readonly() so cross-backend code can use a single check.
Return Type
Boolean
Request Parameters
- prefix (optional)
Sample Case
from ebrains_drive import BucketApiClient
client = BucketApiClient(token="ey...")
bucket = client.buckets.get_bucket("existing_collab_name")
# shows all files (flat, no directory grouping)
all_files = [f for f in bucket.ls()]
# shows all files whose name begins with path/to/my/files
my_files = [f for f in bucket.ls(prefix="path/to/my/files")]For a tree-style view that groups objects by / boundaries, use
Bucket Directory instead.
Return Type
An Iterator of DataproxyFile Objects
Exceptions
- Unauthorized
The data-proxy backend is a flat object store, but BucketDir exposes
the same prefix-as-directory tree view that the DataProxy GUI uses, so
the same code that works against SeafDir can target buckets too.
BucketDir.ls() issues a request with the native delimiter parameter
so the server returns grouped sub-prefixes — efficient even for large
buckets.
Request Parameters
- path (
"/"returns the bucket root)
Sample Case
bucket = client.buckets.get_bucket("existing_collab_name")
root = bucket.get_dir("/") # bucket root
nested = bucket.get_dir("/path/inside") # arbitrary prefix
deeper = root.get_dir("path").get_dir("inside") # chained navigationNo HTTP round-trip is made — listing happens lazily.
Return Type
A BucketDir Object
Request Parameters
- recursive (keyword-only, default
False)False— yield only the immediate children of this prefix. Each entry is either aBucketDir(a sub-prefix) or aDataproxyFile(an object directly under this prefix).True— yield everyDataproxyFileunder this prefix at any depth.
Sample Case
root = bucket.get_dir("/")
# Tree view: subdirectories + files at this level
for entry in root.ls():
print(entry)
# BucketDir[bucket=foo, path=/dir1]
# BucketDir[bucket=foo, path=/dir2]
# DataproxyFile[bucket=foo, path=top.txt, size=12]
# Or flatten everything under this prefix:
all_files = list(root.ls(recursive=True))Return Type
An iterator of BucketDir and DataproxyFile (or just DataproxyFile
when recursive=True).
Request Parameters
- name
Sample Case
root = bucket.get_dir("/")
new_dir = root.mkdir("future")Note: object-storage directories exist purely by convention — they
spring into existence when an object is uploaded under that prefix.
mkdir therefore performs no HTTP request and changes no
server-side state; it simply returns a new BucketDir for the named
sub-prefix.
Return Type
A BucketDir Object
Request Parameters
- recursive (keyword-only, required; must be
Trueto perform the deletion)
Sample Case
bucket.get_dir("/sub").delete(recursive=True)This deletes every object under the prefix. Calling delete() without
recursive=True raises OperationError — the explicit opt-in is
deliberate because there is no undo.
Return Type
None
Exceptions
OperationError— when called withoutrecursive=True.
Note, if request_access is set to True, this method may require user interaction.
Request Parameters
- dataset_id
- request_access (optional, default
False)
Sample Case
from ebrains_drive import BucketApiClient
client = BucketApiClient(token="ey...")
bucket = client.buckets.get_dataset("dataset_id")
Return Type A Bucket Object
Exceptions
- Unauthorized (if request_access is not set)
Files in buckets are objects in a flat namespace. Names commonly contain
/ to encode a directory-like structure; see
Bucket Directory for a tree-style API over that
convention.
Request Parameters
- filename
Sample Case
from ebrains_drive import BucketApiClient
client = BucketApiClient(token="ey...")
bucket = client.buckets.get_bucket("existing_collab_name")
# OR
bucket = client.buckets.get_dataset("dataset_id")
file_handle = bucket.get_file("filename")Return Type
A File Object
Exceptions
- Unauthorized
- DoesNotExist
Request Parameters
- progress (keyword-only, default
False) — whenTrue, stream the download and display atqdmprogress bar.
Sample Case
from ebrains_drive import BucketApiClient
client = BucketApiClient(token="ey...")
bucket = client.buckets.get_bucket("existing_collab_name")
# OR
bucket = client.buckets.get_dataset("dataset_id")
file_handle = bucket.get_file("filename")
file_content = file_handle.get_content()
file_content = file_handle.get_content(progress=True) # with progress barSeafFile.get_content(progress=...) exists with the same signature on
the Drive backend.
Return Type
bytes
Exceptions
- Unauthorized
- DoesNotExist
Bucket.upload is polymorphic — its first argument may be a path
(str/PathLike) or a file-like object. Bucket.upload_local_file
is the strict path-based variant (mirrors SeafDir.upload_local_file).
Request Parameters
upload:
- filelike — file-like object,
bytes, or path to a local file - filename — name to store the file under in the bucket
**kwargs— passed through to the underlying request (e.g.headers=)
upload_local_file:
- filepath — path to the local file
- name (optional) — destination object name; defaults to the basename
of
filepath - overwrite (default
False) — whenFalse, raiseFileExistsErrorif an object with that name already exists
Sample Case
from ebrains_drive import BucketApiClient
client = BucketApiClient(token="ey...")
bucket = client.buckets.get_bucket("existing_collab_name")
# Path or file-like — both work
bucket.upload("/home/jovyan/test.txt", "test/foobar.txt")
from io import StringIO
fh = StringIO("hello world")
fh.seek(0)
bucket.upload(fh, "test/foobar.txt")
# Path-only variant with overwrite control
bucket.upload_local_file("/home/jovyan/test.txt", "test/foobar.txt", overwrite=True)Return Type
None
Exceptions
- Unauthorized
FileExistsErrorfromupload_local_file(..., overwrite=False)when the destination name is already used.
Server-side copy — no client-side data transfer. Uses the data-proxy's
native PUT /v1/buckets/{name}/{object}/copy endpoint.
Request Parameters (at least one must be supplied)
- dst_name — destination object name (within the same bucket, unless
dst_bucketis also set) - dst_bucket (keyword-only) — destination bucket name
Sample Case
bucket = client.buckets.get_bucket("existing_collab_name")
f = bucket.get_file("report.csv")
# Copy within the same bucket under a new name
f.copy_to(dst_name="report-backup.csv")
# Copy to a different bucket (same name)
f.copy_to(dst_bucket="archive-bucket")
# Copy to a different bucket under a new name
f.copy_to(dst_name="report-archived.csv", dst_bucket="archive-bucket")Return Type
bool — True on success.
Exceptions
ValueError— when neitherdst_namenordst_bucketis supplied.- Unauthorized
Server-side rename within the same bucket — no client-side data transfer.
Uses the data-proxy's native PATCH /v1/buckets/{name}/{object} endpoint.
The data-proxy accepts the rename and completes it in the background, and
reports a rename that fails only by email. So before the request, rename
checks that the object still exists and, unless overwrite is set, that no
object already has the new name. The object can be listed under both names
until the rename completes.
Request Parameters
- newname — new object name, from the root of the bucket. A name without a
folder, such as
"report.csv", moves the object to the root. - overwrite (keyword-only, default
False) — whenFalse, raiseFileExistsErrorif an object namednewnameexists. WhenTrue, the rename replaces that object. - send_success_email (keyword-only, default
False) — whether the data-proxy emails the user when the rename completes. It emails the user about a failed rename either way.
Sample Case
bucket = client.buckets.get_bucket("existing_collab_name")
f = bucket.get_file("reports/report.csv")
# Rename within the same folder
f.rename("reports/report-2026.csv")
# Move to another folder, replacing an object of that name
f.rename("archive/report-2026.csv", overwrite=True)Return Type
bool — True once the data-proxy has accepted the rename. The object's
name is updated to newname.
Exceptions
DoesNotExist— the object is no longer in the bucket.FileExistsError— an object namednewnameexists andoverwriteisFalse.- Unauthorized
Request Parameters
None
Sample Case
from ebrains_drive import BucketApiClient
client = BucketApiClient(token="ey...")
bucket = client.buckets.get_bucket("existing_collab_name")
file_handle = bucket.get_file("filename")
file_handle.delete()Return Type
None
Exceptions
- Unauthorized
- DoesNotExist
- AssertionError
BucketApiClient.file exposes a URL-resolution helper analogous to
DriveApiClient.file. Pass a data-proxy URL and get back a
DataproxyFile:
client = BucketApiClient(token="ey...")
f = client.file.get_file_by_url(
"https://data-proxy.ebrains.eu/api/v1/buckets/my-bucket/path/to/file.txt"
)
# Dataset URLs are supported too
f = client.file.get_file_by_url(
"https://data-proxy.ebrains.eu/api/v1/datasets/<dataset_id>/path/to/file"
)Return Type
A DataproxyFile.
Exceptions
ValueError— when the URL does not match the expected data-proxy format.
The Drive (Seafile) and Bucket (Data-Proxy) clients share a harmonised surface so the same code can target either backend.
ebrains_drive.base defines four runtime-checkable typing.Protocols
that both backends satisfy structurally (no inheritance required):
StorageClient—DriveApiClient,BucketApiClientContainerManager—Repos,BucketsContainer—Repo,BucketStorageObject—SeafFile,DataproxyFile
from ebrains_drive.base import Container, StorageObject
def archive(container: Container, name: str) -> bytes:
obj: StorageObject = container.get_file(name)
return obj.get_content()
# Works against either backend
archive(repo, "/notes.md")
archive(bucket, "notes.md")
# isinstance() against the Protocol works at runtime
assert isinstance(bucket, Container)| concept | Drive | Bucket |
|---|---|---|
| factory | connect(target="drive") |
connect(target="bucket") |
| manager | client.repos |
client.buckets |
| list | client.repos.list() / list_repos() |
client.buckets.list() / list_buckets() |
| get | client.repos.get(id) / get_repo(id) |
client.buckets.get(name) / get_bucket(name) |
| create | client.repos.create(name) / create_repo(name) |
client.buckets.create(name) / create_bucket(name) |
| delete by name | client.repos.delete(id) |
client.buckets.delete(name) / delete_bucket(name) |
| container delete | repo.delete() |
bucket.delete() |
| readonly check | repo.is_readonly() |
bucket.is_readonly() |
| upload (container shortcut) | repo.upload(filelike_or_path, name) / repo.upload_local_file(path) |
bucket.upload(filelike_or_path, name) / bucket.upload_local_file(path) |
| directory view | repo.get_dir("/") → SeafDir |
bucket.get_dir("/") → BucketDir |
| download w/ progress | file.get_content(progress=True) |
file.get_content(progress=True) |
| server-side copy | file.copy_to(dst_dir) / copyTo(...) |
file.copy_to(dst_name=..., dst_bucket=...) |
| rename | file.rename(newname) |
file.rename(newname) |
| URL helper | client.file.get_file_by_url(url) |
client.file.get_file_by_url(url) |
The legacy BucketApiClient.create_new / BucketApiClient.delete_bucket
methods still work but emit a DeprecationWarning pointing to the
canonical client.buckets.create_bucket / client.buckets.delete_bucket
calls.