sites.sdk.sites package

Submodules

sites.sdk.sites.authenticator module

class sites.sdk.sites.authenticator.Authenticator(username: str | None = None, password: str | None = None, token: str | None = None)

An Authenticator instance is an object that manages user authentication against the ECMWF OpenIDConnect authentication system, or simple site unique tokens.

An Authenticator instance requires either credentials or a sites authentication token.

Example usage:

# authenticating using your credentials:

authenticator = Authenticator.from_credentials(username=’username’, password=’password’)

# authentication using the site authentication bearer token:

my_token_authenticator = Authenticator.from_token(token=’c1268a9b07068a9b07…07c8797c7d6d18b69b5b557c1d007dfa’)

# authentication using obfuscated credentials from a previous instance of Authenticator:

my_credentials_base64 = authenticator.get_credentials_base64()

my_authenticator_base64 = Authenticator.from_credentials_base64(credentials=my_credentials_base64)

# authenticator from environment variables, either plain string or base64 encoded

export SITES_AUTHENTICATOR_USERNAME=”username”

export SITES_AUTHENTICATOR_PASSWORD=”cGFzc3dvcmQK” # base64 encoded

authenticator = Authenticator.from_environment_variables()

Arguments to instantiate a Authenticator:

Param:

username: str - a username

Param:

password: str - a password

Param:

token: str - a Bearer unique site token

Raise:

Exception if http_code is not in 20x

classmethod from_credentials(username: str, password: str) → Authenticator

Creates an instance of Authenticator from user credentials username / password.

Param:

username: str - a username

Param:

password: str - a password

Returns:

an instance of Authenticator

classmethod from_credentials_base64(credentials: str) → Authenticator

Creates an instance of Authenticator from user credentials encoded in base64.

Param:

credentials:

Returns:

an instance of Authenticator

Raise:

Exception if credentials are not provided or have an invalid format or if http_code is not in 20x

classmethod from_environment_variables() → Authenticator

Creates an instance of Authenticator from values defined in Environment Variables.

Expects 2 Environment Variables SITES_AUTHENTICATOR_USERNAME and SITES_AUTHENTICATOR_PASSWORD.

These variables can be defined with plain strings or base64 encoded values. When SITES_AUTHENTICATOR_ENCODING=base64 is set the values are always decoded, bypassing the auto-detection heuristic. This is the recommended approach when the credential values are known to be base64-encoded, as the auto-detection may not be reliable for all passwords.

Returns:

an instance of Authenticator

classmethod from_token(token: str) → Authenticator

Creates an instance of Authenticator from a Bearer Token. Check the Site configuration to get it.

Param:

token: str - the Site unique generated token

Returns:

an instance of Authenticator

get_authorization_header() → Dict[str, str]

Returns the authorization header in the format: ‘Authorization: Bearer eyJhbGciOiJIUz.eyJzdWIiOiIx.SflKxwRJc’

Returns:

str - the authorization header

get_credentials_base64() → str

Returns the credentials in base64 format

Returns:

format is {username.password} in base64

get_token() → str | None | Any

Returns the JWT token or the Site unique Bearer token, depending on the authentication being used.

Returns:

the JWT token

Raise:

Exception if http_code is not in 20x

sites.sdk.sites.site module

class sites.sdk.sites.site.Site(name: str | None = None, space: str | None = None, owner: str | None = None, template: str | None = None, config: Configuration = <sites.sdk.common.configuration.Configuration object>)

A site URL is composed by the username and a site name: https://sites.ecmwf.int/username/site_name

In some cases a site can be configured instead with a custom space instead of using the username, e.g., https://sites.ecmwf.int/custom_space/site_name

This class is used to represent either of these cases.

A Site instance requires a space a name and an Authenticator.

Example usage:

# creating an authenticator

authenticator = Authenticator.from_token(‘c1268a9b07068a9b07…07c8797c7d6d18b69b5b557c1d007dfa’)

# instantiating a site with username space

site = Site(space=’owner’, name=’site-name’, authenticator=authenticator)

# instantiating a site with a custom space

site = Site(space=’custom_space’, name=’site-name’, authenticator=authenticator)

# instantiating a site in stage:

site = Site(space=’user’, name=’name’, authenticator=authenticator, sites_base_url=’https://sites-stage.ecmwf.int’)

Arguments to instantiate a Site:

Param:

space: str - the username or custom space of the site

Param:

name: str - the site name

Param:

config: Configuration - the configuration file

classmethod from_dict(dic: ~typing.Dict[str, ~typing.Any], config: ~sites.sdk.common.configuration.Configuration = <sites.sdk.common.configuration.Configuration object>) → Site

Creates an instance of Site from a dictionary.

Param:

dic: dict - a dictionary to retrieve data from

Param:

config: Configuration - a configuration object

Returns:

an instance of Site

classmethod from_owner_and_name(owner: str, name: str, config: Configuration = <sites.sdk.common.configuration.Configuration object>) → Site

Creates an instance of Site from the site owner and name.

Deprecated since version Use: from_space_and_name() instead.

Param:

owner: str - the site owner

Param:

name: str - the site name

Param:

config: Configuration - optional - a configuration object

Returns:

an instance of Site

classmethod from_space_and_name(space: str, name: str, config: Configuration = <sites.sdk.common.configuration.Configuration object>) → Site

Creates an instance of Site from a space and site name.

Param:

space: str - the site space

Param:

name: str - the site name

Param:

config: Configuration - optional - a configuration object

Returns:

an instance of Site

classmethod from_template(template: str, config: Configuration = <sites.sdk.common.configuration.Configuration object>) → Site

Creates an instance of Site from a site template.

Param:

template: str - a site template, e.g. iver, ecbox, repository

Returns:

an instance of Site

get_base_url() → str

Returns the Sites Hub base url

Returns:

str - the base url

get_config() → Configuration

Returns the current Configuration object

Returns:

an instance of Configuration

get_content_manager(authenticator: AuthorizationBuilder | None = None) → SiteContentManagement

Returns an instance of current SiteContentManagement to manage the current Sites’ content

Param:

authenticator: AuthorizationBuilder - the authenticator to use

Returns:

an instance of SiteContentManagement

get_path() → str

Returns the current Sites’ path

Returns:

str - the path in the form ‘space/name’

load_from_server(authenticator: AuthorizationBuilder | None = None) → None

This method loads the current site configuration by requesting its information from the API

Param:

authenticator: AuthorizationBuilder - the authenticator to use

sites.sdk.sites.site_content module

class sites.sdk.sites.site_content.SiteContentManagement(site: Site, authenticator: AuthorizationBuilder | None = None)

Client to manage sites content

Arguments to instantiate a SiteContentManagement:

Param:

site: Site - the site to manage the content

Param:

authenticator: AuthorizationBuilder - the authenticator to be used by default

backup_token_history() → Dict[str, Any]

Create a backup of all token history entries.

Creates a gzipped JSON backup of all token history entries stored on the server.

Returns:

returns a JSON with backup metadata including timestamp, row count, and SHA256 checksum

checksum(remote_path: str) → Dict[str, Any]

Retrieve the SHA-256 checksum and metadata for a remote file without downloading it.

Sends a HEAD request with Want-Repr-Digest: sha-256 and returns metadata from the response headers.

Parameters:

remote_path – str - the remote file path

Returns:

dict with keys content_length, content_type, last_modified, etag, checksum (base64-encoded SHA-256 or None for empty files)

Raises:

• FileNotFoundError – if the file does not exist (404)

• ApiCallException – on any other non-success status

clone(to_site: Site | None = None, authenticator: AuthorizationBuilder | None = None, cleanup_after: bool = False, ignore_errors: bool = False, local_path: str = '/tmp/sites/') → None

Clones all the content of the site into the file system first and uploads all the content to another site, if a to_site is provided.

Parameters:

• to_site – Site, optional - the site to clone to

• authenticator – AuthorizationBuilder - optional - the authenticator for the to_site

• cleanup_after – bool, optional - if the local file system clone should be clean after completion (Default to False)

• ignore_errors – bool, optional - if errors should be ignored (Default to False)

create_symlink(path: str, target_path: str, force: bool = False) → Dict[str, Any]

Create a symlink to a file or directory remotely on your website.

Param:

path: str - the path of the symlink to create

Param:

target_path: str - the path of the symlink target on the remote file system

Param:

force: bool - if the operation should be forced (in case a symlink already exists)

Returns:

returns a JSON with the generated symlink

decode_token(token: str) → Dict[str, Any]

Decode a branca token used to access this websites’ REST API.

Param:

token: str - the token to decode

Returns:

returns a JSON with the token data

delete(remote_path: str = '', match_pattern: str = '', recursive: bool = False) → Dict[str, Any]

Deletes content from the site.

Param:

remote_path: str, optional - the remote path to clean

Param:

match_pattern: str, optional - a glob pattern to run as a match pattern for files

Param:

recursive: bool, optional - if the search should be recursive (Default to False)

Returns:

a JSON with the result of the deletion. On 200 this is a standard ApiResponse. On 207 Multi-Status (partial failure when using match_pattern) the dict contains deleted (list of removed paths) and failed (list of dicts with path and reason keys).

delete_symlink(path: str, force: bool = False) → Dict[str, Any]

Delete a symbolic link on your website.

Param:

path: str - the path of the symlink to delete

Param:

force: bool - if the operation should be forced (Default to False)

Returns:

returns a JSON with the result of the deletion

Raises:

• RuntimeError – if path is empty

• FileNotFoundError – if the symlink does not exist (404)

• ApiCallException – on any other non-success status

delete_token_history(token_hash: str) → Dict[str, Any]

Delete a token history entry by its hash.

Param:

token_hash: str - the SHA256 hash of the token to delete (64-character hex string)

Returns:

returns a JSON with the result of the deletion

download(remote_path: str = '', match_pattern: str = '', verify: bool = False, archive: ArchiveFormat | None = None) → Dict[str, Any] | bytes

Downloads content from the site. If using a regex or doing recursive searched list_files should be set to True as this method can only download a file at a time.

Param:

remote_path: str - the remote path of the file or directory

Param:

match_pattern: str - a glob expression to run as a match pattern for files

Param:

verify: bool - if True, request and verify Repr-Digest when available

Param:

archive: ArchiveFormat, optional - if provided, the response will be a compressed archive in the specified format (zip, tar, or tgz). The server does not return Repr-Digest for archive responses, so verify is ignored when archive is set.

Raises:

ChecksumMismatchError – if verify=True and the local digest does not match the server-reported digest

Returns:

returns a JSON with the list of files or the content of the file

download_with_checksum(remote_path: str = '', match_pattern: str = '', verify: bool = False) → Dict[str, Any]

Download content from the site and optionally verify its SHA-256 checksum.

Deprecated since version next: release Use download() with verify=True. This method will be removed in the next release.

When verify is True and the server returns a Repr-Digest header, the downloaded content is hashed locally and compared against the server digest. A ChecksumMismatchError is raised on mismatch.

Parameters:

• remote_path – str - the remote path of the file

• match_pattern – str - a glob expression to match files

• verify – bool - if True, verify the checksum after download (default False)

Returns:

dict with keys content (bytes or dict), checksum (str or None), etag (str or None), verified (bool)

Raises:

• ChecksumMismatchError – if verify is True and the local digest does not match the server-reported digest

• FileNotFoundError – if the file does not exist (404)

• ApiCallException – on any other non-success status

generate_signed_url(target_url: str, duration: int, permission: SiteAccessPermission = SiteAccessPermission.Read) → Dict[str, Any]

Generate a Signed Url to access a specified resource on a specified Site.

Param:

url: str - a url to sign

Param:

duration: int - the duration of the token in seconds

Param:

permission: SiteTokenPermission - the permission of the token, Read or ReadWrite

Returns:

returns a JSON with the generated token

generate_token(identifier: str, duration: int, permission: SiteAccessPermission = SiteAccessPermission.Read) → Dict[str, Any]

Generate a branca token to be used to access this websites’ REST API.

Param:

identifier: str - an identifier that will be used as username for audit purposes

Param:

duration: int - the duration of the token in seconds

Param:

permission: SiteTokenPermission - the permission of the token, Read or ReadWrite

Returns:

returns a JSON with the generated token

get_symlink(path: str) → Dict[str, Any]

Get information about a symbolic link on your website.

Param:

path: str - the path of the symlink to query

Returns:

returns a JSON with the symlink information (including target)

Raises:

• RuntimeError – if path is empty

• FileNotFoundError – if the symlink does not exist (404)

• ApiCallException – on any other non-success status

get_version() → Dict[str, Any]

Returns the version of the site API.

Returns:

returns a JSON with the site version, e.g. {'version': '2.0.0'}

Raises:

ApiCallException – on any non-success status

health() → Dict[str, Any]

Returns the health status of the site.

Returns:

returns a JSON with the site health status

list(remote_path: str | None = '/', match_pattern: str | None = '*', file_type: FileType | None = None, recursive: bool = False, limit: int = 100, continuation_token: str | None = None, version: int = 0) → Dict[str, Any]

Please note that this function has been abstracted in order to allow the use of both versions of the REST API, and now defaults to Configuration.API_VERSION.

Param:

recursive: bool - recurse all directories

Param:

file_type: FileType (f, d) - filter by file type, file ‘f’ or directory ‘d’

Param:

remote_path: str - the remote path of the file or directory (resource) to list

Param:

match_pattern: str - a glob expression to run as a match pattern for files

Param:

limit: int - the max number of items to retrieve

Param:

continuation_token: str - the token retrieved from a previous listing, in order to retrieve the remaining results

Param:

version: int - the version of the api to use

Returns:

returns a JSON with the list of files or the content of the file

list_v1(remote_path: str | None = '/', match_pattern: str | None = '*', file_type: FileType | None = None, recursive: bool = False) → Dict[str, Any]

Lists content from a site using version 1 of the REST API.

See list().

Param:

recursive: bool - recurse all directories

Param:

file_type: FileType (f, d) - filter by file type, file ‘f’ or directory ‘d’

Param:

remote_path: str - the remote path of the file or directory (resource) to list

Param:

match_pattern: str - a glob expression to run as a match pattern for files

Returns:

returns a JSON with the list of files or the content of the file

list_v2(remote_path: str | None = '/', match_pattern: str | None = '*', file_type: FileType | None = None, recursive: bool = False, limit: int = 100, continuation_token: str | None = None) → Dict[str, Any]

Lists content from a site using version 2 of the REST API.

See list().

Param:

recursive: bool - recurse all directories

Param:

file_type: FileType (f, d) - filter by file type, file ‘f’ or directory ‘d’

Param:

remote_path: str - the remote path of the file or directory (resource) to list

Param:

match_pattern: str - a glob expression to run as a match pattern for files

Param:

limit: int - the max number of items to retrieve

Param:

continuation_token: str - the token retrieved from a previous listing, in order to retrieve the remaining results

Param:

version: int - the version of the api to use

Returns:

returns a JSON with the list of files or the content of the file

list_walk(remote_path: str | None = '/', match_pattern: str | None = '*', file_type: FileType | None = None, recursive: bool = False, limit: int = 100) → Iterator[Any]

Lists content from a site.

Example:

for path in list_walk(recursive=True, match=’*’):

print(path)

Param:

recursive: bool - recurse all directories

Param:

file_type: FileType (f, d) - filter by file type, file ‘f’ or directory ‘d’

Param:

remote_path: str - the remote path of the file or directory (resource) to list

Param:

match_pattern: str - a glob expression to run as a match pattern for files

Returns:

returns a JSON with the list of files or the content of the file

publish() → Dict[str, Any]

restore_token_history() → Dict[str, Any]

Restore token history from the latest backup.

This is a destructive operation: all current token history entries are deleted and replaced with the backup data.

Returns:

returns a JSON with restore metadata including rows restored count

revoke_token(token: str | None = None, token_hash: str | None = None) → Dict[str, Any]

Revoke a branca token by either the token string or its hash.

Provide either ‘token’ or ‘token_hash’ to identify which token to revoke. Do not provide both.

Param:

token: str, optional - the token string to revoke

Param:

token_hash: str, optional - the SHA256 hash of the token to revoke (64-character hex string)

Returns:

returns a JSON with the result of the revocation

token_history(identifier: str | None = None, permission: SiteAccessPermission | None = None, expiring_within: int | None = None, include_expired: bool | None = None) → Dict[str, Any]

Retrieve token usage history with optional filtering.

Returns a list of token usage history entries. Each unique identifier+permission pair is tracked with its most recent usage information only. By default, expired tokens are excluded unless explicitly requested with include_expired=True.

Param:

identifier: str, optional - filter by token identifier

Param:

permission: SiteAccessPermission, optional - filter by token permission (Read or ReadWrite)

Param:

expiring_within: int, optional - filter tokens expiring within N seconds (e.g., 604800 for 7 days)

Param:

include_expired: bool, optional - include expired tokens in results (default False)

Returns:

returns a JSON with the token history list

upload(local_path: str, match_pattern: str = '*', remote_path: str | None = None, recursive: bool = False, ignore_errors: bool = False, if_exists_rename_to: str | None = None, if_exists_backup: bool = False, upload_zero_byte_files: bool = False) → List[Dict[str, Any]]

Uploads content to a site.

Param:

recursive: bool - if search should be recursive (Default to False)

Param:

local_path: str - the local path to upload, if this is a directory the whole directory will be uploaded

Param:

match_pattern: str, optional - a regex to match the files in local_path to be uploaded

Param:

remote_path: str, optional - the remote path to upload to, can be a new folder, etc.

Param:

ignore_errors: bool, optional - if errors should be ignored (Default to False)

Param:

if_exists_rename_to: str, optional - rename a file to this if it exists

Param:

if_exists_backup: bool, optional - backs-up a file if it exists (Default to False)

Param:

upload_zero_byte_files: bool, optional - to upload empty files (Default to False)

Returns:

list[dict] - one result dict per uploaded file. Each dict contains at minimum status and message keys. A checksum key is included when the server returns a Repr-Digest header.

verify_signed_url(url: str) → Dict[str, Any]

Verify if a provided Signed Url is valid to access a specified resource on specified Site.

The API returns 204 No Content on success (empty body), so this method returns a simple {'status': 'ok'} dict in that case.

Param:

url: str - the url to verify

Returns:

returns a dict — {'status': 'ok'} on 204 success, or the parsed JSON body otherwise

sites.sdk.sites.utils module

class sites.sdk.sites.utils.ArchiveFormat(*values)

Represents the supported archive formats for downloading compressed content.

Supported values: ‘zip’, ‘tar’, ‘tgz’

TAR = 'tar'

TGZ = 'tgz'

ZIP = 'zip'

static convert(value: str, default: str | None = None) → str | None

Will return the current value if is a valid one, otherwise defaults to None @param value: the value to convert @param default: the default value to return if value is not valid @return: the value if valid, otherwise default

static is_valid(value: str) → bool

class sites.sdk.sites.utils.ContextRoot(*values)

Represents the site root context: ‘Root Context’ or ‘Site Context’ Depending on the custom application, one can select if this will be served from the ‘Root context’ -> ‘/’ or from the ‘Site Context’ -> ‘/space/name/’

RootContext = 'root_context'

SiteContext = 'site_context'

static convert(value: str, default: str | None = None) → str | None

Will return the current value if is a valid one, otherwise defaults to None @param value: the value to convert @param default: the default value to return if value is not valid @return: the value if valid, otherwise default

static is_valid(value: str) → bool

class sites.sdk.sites.utils.FileType(*values)

Represents the file types: ‘File’ or ‘Directory’ When listing files from a specific site, one might want to list depending on the file type

DIR = 'd'

FILE = 'f'

static convert(value: str, default: str | None = None) → str | None

Will return the current value if it is a valid one, otherwise defaults to None @param value: the value to convert @param default: the default value to return if value is not valid @return: the value if valid, otherwise the default (if not provided defaults to None)

static is_valid(value: str) → bool

class sites.sdk.sites.utils.ProxyType(*values)

Represents the type of proxy to serve applications: ‘ProxyPass’ or ‘FastCGI’ On custom applications users can serve their own image using either ‘ProxyPass’ or ‘FastCGI’

FastCGI = 'fastcgi'

ProxyPass = 'proxy'

static convert(value: str, default: str | None = None) → str | None

Will return the current value if is a valid one, otherwise defaults to None @param value: the value to convert @param default: the default value to return if value is not valid @return: the value if valid, otherwise default

static is_valid(value: str) → bool

class sites.sdk.sites.utils.SiteAccessPermission(*values)

Represents the possible types of permissions: ‘Read’ or ‘ReadWrite’

Where ‘Read’ is read only access, and ‘ReadWrite’ is read/write access.

ReadWrite is Administrator level permissions which allow users to generate tokens as well!

Read = 'r'

ReadWrite = 'rw'

static convert(value: str, default: str | None = None) → str | None

Will return the current value if is a valid one, otherwise defaults to None @param value: the value to convert @param default: the default value to return if value is not valid @return: the value if valid, otherwise default

static is_valid(value: str) → bool

class sites.sdk.sites.utils.SiteType(*values)

Represents the possible types of sites: ‘Private’ or ‘Public’ ‘Public’ sites are available to anyone without authorization ‘Private’ sites are available only to specific users or groups of users, they must be authenticated and match the ACL

Private = 'private'

Public = 'public'

static convert(value: str, default: str | None = None) → str | None

Will return the current value if is a valid one, otherwise defaults to None @param value: the value to convert @param default: the default value to return if value is not valid @return: the value if valid, otherwise default

static is_valid(value: str) → bool