Files

albert.collections.files

AlbertSession

AlbertSession(
    *,
    base_url: str,
    token: str | None = None,
    client_credentials: ClientCredentials | None = None,
    retries: int | None = None,
)

Bases: Session

A session that has a base URL, which is prefixed to all request URLs.

Parameters:

Name	Type	Description	Default
base_url	str	The base URL to prefix to all requests. (e.g., "https://sandbox.albertinvent.com")	required
retries	int	The number of retries for failed requests. Defaults to 3.	None
client_credentials	ClientCredentials | None	The client credentials for programmatic authentication. Optional if token is provided.	None
token	str | None	The JWT token for authentication. Optional if client credentials are provided.	None

Methods:

Name	Description
request

Source code in src/albert/session.py

def __init__(
    self,
    *,
    base_url: str,
    token: str | None = None,
    client_credentials: ClientCredentials | None = None,
    retries: int | None = None,
):
    super().__init__()
    self.base_url = base_url
    self.headers.update(
        {
            "Content-Type": "application/json",
            "Accept": "application/json",
            "User-Agent": f"albert-SDK V.{albert.__version__}",
        }
    )

    if token is None and client_credentials is None:
        raise ValueError("Either client credentials or token must be specified.")

    self._provided_token = token
    self._token_manager = (
        TokenManager(base_url, client_credentials) if client_credentials is not None else None
    )

    # Set up retry logic
    retries = retries if retries is not None else 3
    retry = Retry(
        total=retries,
        read=retries,
        connect=retries,
        backoff_factor=0.3,
        status_forcelist=(500, 502, 503, 504, 403),
        raise_on_status=False,
    )
    adapter = HTTPAdapter(max_retries=retry)
    self.mount("http://", adapter)
    self.mount("https://", adapter)

base_url instance-attribute

base_url = base_url

request

request(
    method: str, path: str, *args, **kwargs
) -> Response

Source code in src/albert/session.py

def request(self, method: str, path: str, *args, **kwargs) -> requests.Response:
    self.headers["Authorization"] = f"Bearer {self._access_token}"
    full_url = urljoin(self.base_url, path) if not path.startswith("http") else path
    with handle_http_errors():
        response = super().request(method, full_url, *args, **kwargs)
        response.raise_for_status()
        return response

BaseCollection

BaseCollection(*, session: AlbertSession)

BaseCollection is the base class for all collection classes.

Parameters:

Name	Type	Description	Default
session	AlbertSession	The Albert API Session instance.	required

Source code in src/albert/collections/base.py

def __init__(self, *, session: AlbertSession):
    self.session = session

session instance-attribute

session = session

FileCategory

Bases: str, Enum

OTHER class-attribute instance-attribute

OTHER = 'Other'

SDS class-attribute instance-attribute

SDS = 'SDS'

FileCollection

FileCollection(*, session: AlbertSession)

Bases: BaseCollection

FileCollection is a collection class for managing File entities in the Albert platform.

Parameters:

Name	Type	Description	Default
session	AlbertSession	The Albert session instance.	required

Methods:

Name	Description
get_by_name	Gets a file by name and namespace.
get_signed_download_url	Get a signed download URL for a file.
get_signed_upload_url	Get a signed upload URL for a file.
sign_and_upload_file	Sign and upload a file to Albert.

Source code in src/albert/collections/files.py

def __init__(self, *, session: AlbertSession):
    """
    Initialize the FileCllection with the provided session.

    Parameters
    ----------
    session : AlbertSession
        The Albert session instance.
    """
    super().__init__(session=session)
    self.base_path = f"/api/{FileCollection._api_version}/files"

base_path instance-attribute

base_path = f'/api/{_api_version}/files'

get_by_name

get_by_name(
    *,
    name: str,
    namespace: FileNamespace,
    generic: bool = False,
) -> FileInfo

Gets a file by name and namespace.

Parameters:

Name	Type	Description	Default
name	str	The Name of the file	required
namespace	FileNamespace	The namespace of the file (e.g. AGENT, BREAKTHROUGH, PIPELINE, PUBLIC, RESULT, SDS)	required
generic	bool	description, by default False	False

Returns:

Type	Description
FileInfo	The file information related to the matching file.

Source code in src/albert/collections/files.py

def get_by_name(
    self,
    *,
    name: str,
    namespace: FileNamespace,
    generic: bool = False,
) -> FileInfo:
    """Gets a file by name and namespace.

    Parameters
    ----------
    name : str
        The Name of the file
    namespace : FileNamespace
        The namespace of the file (e.g. AGENT, BREAKTHROUGH, PIPELINE, PUBLIC, RESULT, SDS)
    generic : bool, optional
        _description_, by default False

    Returns
    -------
    FileInfo
        The file information related to the matching file.
    """
    params = {
        "name": name,
        "namespace": namespace,
        "generic": json.dumps(generic),
    }
    response = self.session.get(f"{self.base_path}/info", params=params)
    return FileInfo(**response.json())

get_signed_download_url

get_signed_download_url(
    *,
    name: str,
    namespace: FileNamespace,
    version_id: str | None = None,
    generic: bool = False,
    category: FileCategory | None = None,
) -> str

Get a signed download URL for a file.

Parameters:

Name	Type	Description	Default
name	str	The Name of the file	required
namespace	FileNamespace	The namespace of the file (e.g. AGENT, BREAKTHROUGH, PIPELINE, PUBLIC, RESULT, SDS)	required
version_id	str | None	The version of the file, by default None	None
category	FileCategory | None	The file category (E.g., SDS, OTHER), by default None	None

Returns:

Type	Description
str	description

Source code in src/albert/collections/files.py

def get_signed_download_url(
    self,
    *,
    name: str,
    namespace: FileNamespace,
    version_id: str | None = None,
    generic: bool = False,
    category: FileCategory | None = None,
) -> str:
    """Get a signed download URL for a file.

    Parameters
    ----------
    name : str
        The Name of the file
    namespace : FileNamespace
        The namespace of the file (e.g. AGENT, BREAKTHROUGH, PIPELINE, PUBLIC, RESULT, SDS)
    version_id : str | None, optional
        The version of the file, by default None
    category : FileCategory | None, optional
        The file category (E.g., SDS, OTHER), by default None

    Returns
    -------
    str
        _description_
    """
    params = {
        "name": name,
        "namespace": namespace,
        "versionId": version_id,
        "generic": json.dumps(generic),
        "category": category,
    }
    response = self.session.get(
        f"{self.base_path}/sign",
        params={k: v for k, v in params.items() if v is not None},
    )
    return response.json()["URL"]

get_signed_upload_url

get_signed_upload_url(
    *,
    name: str,
    namespace: FileNamespace,
    content_type: str,
    generic: bool = False,
    category: FileCategory | None = None,
) -> str

Get a signed upload URL for a file.

Parameters:

Name	Type	Description	Default
name	str	The Name of the file	required
namespace	FileNamespace	The namespace of the file (e.g. AGENT, BREAKTHROUGH, PIPELINE, PUBLIC, RESULT, SDS)	required
content_type	str	The content type of the file	required
category	FileCategory | None	The File category (E.g., SDS, OTHER), by default None	None

Returns:

Type	Description
str	description

Source code in src/albert/collections/files.py

def get_signed_upload_url(
    self,
    *,
    name: str,
    namespace: FileNamespace,
    content_type: str,
    generic: bool = False,
    category: FileCategory | None = None,
) -> str:
    """Get a signed upload URL for a file.

    Parameters
    ----------
    name : str
        The Name of the file
    namespace : FileNamespace
        The namespace of the file (e.g. AGENT, BREAKTHROUGH, PIPELINE, PUBLIC, RESULT, SDS)
    content_type : str
        The content type of the file
    category : FileCategory | None, optional
        The File category (E.g., SDS, OTHER), by default None

    Returns
    -------
    str
        _description_
    """
    params = {"generic": json.dumps(generic)}

    post_body = SignURLPOST(
        files=[
            SignURLPOSTFile(
                name=name,
                namespace=namespace,
                content_type=content_type,
                category=category,
            )
        ]
    )

    response = self.session.post(
        f"{self.base_path}/sign",
        json=post_body.model_dump(by_alias=True, exclude_unset=True, mode="json"),
        params=params,
    )
    return response.json()[0]["URL"]

sign_and_upload_file

sign_and_upload_file(
    data: IO,
    name: str,
    namespace: FileNamespace,
    content_type: str,
    generic: bool = False,
    category: FileCategory | None = None,
) -> None

Sign and upload a file to Albert.

Parameters:

Name	Type	Description	Default
data	IO	The file data	required
name	str	The name of the file	required
namespace	FileNamespace	The File Namespace (e.g., AGENT, BREAKTHROUGH, PIPELINE, PUBLIC, RESULT, SDS)	required
content_type	str	The content type of the file	required
category	FileCategory | None	The category of the file (E.g., SDS, OTHER), by default None	None

Source code in src/albert/collections/files.py

def sign_and_upload_file(
    self,
    data: IO,
    name: str,
    namespace: FileNamespace,
    content_type: str,
    generic: bool = False,
    category: FileCategory | None = None,
) -> None:
    """Sign and upload a file to Albert.

    Parameters
    ----------
    data : IO
        The file data
    name : str
        The name of the file
    namespace : FileNamespace
        The File Namespace (e.g., AGENT, BREAKTHROUGH, PIPELINE, PUBLIC, RESULT, SDS)
    content_type : str
        The content type of the file
    category : FileCategory | None, optional
        The category of the file (E.g., SDS, OTHER), by default None
    """
    upload_url = self.get_signed_upload_url(
        name=name,
        namespace=namespace,
        content_type=content_type,
        generic=generic,
        category=category,
    )
    requests.put(upload_url, data=data, headers={"Content-Type": content_type})

FileInfo

Bases: BaseAlbertModel

content_type class-attribute instance-attribute

content_type: str = Field(..., alias='contentType')

etag instance-attribute

etag: str

last_modified class-attribute instance-attribute

last_modified: datetime = Field(..., alias='lastModified')

metadata class-attribute instance-attribute

metadata: list[dict[str, str]] = Field(
    ..., default_factory=list
)

name instance-attribute

name: str

namespace class-attribute instance-attribute

namespace: FileNamespace | None = Field(default=None)

size instance-attribute

size: int

FileNamespace

Bases: str, Enum

AGENT class-attribute instance-attribute

AGENT = 'agent'

BREAKTHROUGH class-attribute instance-attribute

BREAKTHROUGH = 'breakthrough'

PIPELINE class-attribute instance-attribute

PIPELINE = 'pipeline'

PUBLIC class-attribute instance-attribute

PUBLIC = 'public'

RESULT class-attribute instance-attribute

RESULT = 'result'

SDS class-attribute instance-attribute

SDS = 'sds'

SignURLPOST

Bases: BaseAlbertModel

files instance-attribute

files: list[SignURLPOSTFile]

SignURLPOSTFile

Bases: BaseAlbertModel

category class-attribute instance-attribute

category: FileCategory | None = Field(default=None)

content_type class-attribute instance-attribute

content_type: str = Field(..., alias='contentType')

metadata class-attribute instance-attribute

metadata: list[dict[str, str]] | None = Field(default=None)

name instance-attribute

name: str

namespace instance-attribute

namespace: FileNamespace

url class-attribute instance-attribute

url: str | None = Field(default=None)