Skip to content

Files

albert.collections.files

AlbertSession 

AlbertSession(
    *,
    base_url: str,
    token: str | None = None,
    auth_manager: AlbertClientCredentials
    | AlbertSSOClient
    | 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 relative request paths (e.g., "https://app.albertinvent.com").

 required
token str | None

A static JWT token for authentication. Ignored if auth_manager is provided.

 None
auth_manager AlbertClientCredentials | AlbertSSOClient

An authentication manager used to dynamically fetch and refresh tokens. If provided, it overrides token.

 None
retries int

The number of automatic retries on failed requests (default is 3).

 None

Methods:

Name Description
request
Source code in src/albert/core/session.py 
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
def __init__(
    self,
    *,
    base_url: str,
    token: str | None = None,
    auth_manager: AlbertClientCredentials | AlbertSSOClient | 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 auth_manager is None:
        raise ValueError("Either `token` or `auth_manager` must be specified.")

    self._auth_manager = auth_manager
    self._provided_token = token

    # 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/core/session.py 
75
76
77
78
79
80
81
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 
21
22
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 
22
23
24
25
26
27
28
29
30
31
32
def __init__(self, *, session: AlbertSession):
    """
    Initialize the FileCollection 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

TODO: 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 
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
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
        TODO: _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

S3 signed URL.
Source code in src/albert/collections/files.py 
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
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
        S3 signed URL.
    """
    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

S3 signed URL.
Source code in src/albert/collections/files.py 
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
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
        S3 signed URL.
    """
    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 
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
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 pydantic-model

Bases: BaseAlbertModel

Show JSON schema: 
{
  "$defs": {
    "FileNamespace": {
      "enum": [
        "agent",
        "breakthrough",
        "pipeline",
        "public",
        "result",
        "sds"
      ],
      "title": "FileNamespace",
      "type": "string"
    }
  },
  "properties": {
    "name": {
      "title": "Name",
      "type": "string"
    },
    "size": {
      "title": "Size",
      "type": "integer"
    },
    "etag": {
      "title": "Etag",
      "type": "string"
    },
    "namespace": {
      "anyOf": [
        {
          "$ref": "#/$defs/FileNamespace"
        },
        {
          "type": "null"
        }
      ],
      "default": null
    },
    "contentType": {
      "title": "Contenttype",
      "type": "string"
    },
    "lastModified": {
      "format": "date-time",
      "title": "Lastmodified",
      "type": "string"
    },
    "metadata": {
      "items": {
        "additionalProperties": {
          "type": "string"
        },
        "type": "object"
      },
      "title": "Metadata",
      "type": "array"
    }
  },
  "required": [
    "name",
    "size",
    "etag",
    "contentType",
    "lastModified"
  ],
  "title": "FileInfo",
  "type": "object"
}

Fields:

content_type pydantic-field 

content_type: str

etag pydantic-field 

etag: str

last_modified pydantic-field 

last_modified: datetime

metadata pydantic-field 

metadata: list[dict[str, str]]

name pydantic-field 

name: str

namespace pydantic-field 

namespace: FileNamespace | None = None

size pydantic-field 

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 pydantic-model

Bases: BaseAlbertModel

Show JSON schema: 
{
  "$defs": {
    "FileCategory": {
      "enum": [
        "SDS",
        "Other"
      ],
      "title": "FileCategory",
      "type": "string"
    },
    "FileNamespace": {
      "enum": [
        "agent",
        "breakthrough",
        "pipeline",
        "public",
        "result",
        "sds"
      ],
      "title": "FileNamespace",
      "type": "string"
    },
    "SignURLPOSTFile": {
      "properties": {
        "name": {
          "title": "Name",
          "type": "string"
        },
        "namespace": {
          "$ref": "#/$defs/FileNamespace"
        },
        "contentType": {
          "title": "Contenttype",
          "type": "string"
        },
        "metadata": {
          "anyOf": [
            {
              "items": {
                "additionalProperties": {
                  "type": "string"
                },
                "type": "object"
              },
              "type": "array"
            },
            {
              "type": "null"
            }
          ],
          "default": null,
          "title": "Metadata"
        },
        "category": {
          "anyOf": [
            {
              "$ref": "#/$defs/FileCategory"
            },
            {
              "type": "null"
            }
          ],
          "default": null
        },
        "url": {
          "anyOf": [
            {
              "type": "string"
            },
            {
              "type": "null"
            }
          ],
          "default": null,
          "title": "Url"
        }
      },
      "required": [
        "name",
        "namespace",
        "contentType"
      ],
      "title": "SignURLPOSTFile",
      "type": "object"
    }
  },
  "properties": {
    "files": {
      "items": {
        "$ref": "#/$defs/SignURLPOSTFile"
      },
      "title": "Files",
      "type": "array"
    }
  },
  "required": [
    "files"
  ],
  "title": "SignURLPOST",
  "type": "object"
}

Fields:

files pydantic-field 

files: list[SignURLPOSTFile]

SignURLPOSTFile pydantic-model

Bases: BaseAlbertModel

Show JSON schema: 
{
  "$defs": {
    "FileCategory": {
      "enum": [
        "SDS",
        "Other"
      ],
      "title": "FileCategory",
      "type": "string"
    },
    "FileNamespace": {
      "enum": [
        "agent",
        "breakthrough",
        "pipeline",
        "public",
        "result",
        "sds"
      ],
      "title": "FileNamespace",
      "type": "string"
    }
  },
  "properties": {
    "name": {
      "title": "Name",
      "type": "string"
    },
    "namespace": {
      "$ref": "#/$defs/FileNamespace"
    },
    "contentType": {
      "title": "Contenttype",
      "type": "string"
    },
    "metadata": {
      "anyOf": [
        {
          "items": {
            "additionalProperties": {
              "type": "string"
            },
            "type": "object"
          },
          "type": "array"
        },
        {
          "type": "null"
        }
      ],
      "default": null,
      "title": "Metadata"
    },
    "category": {
      "anyOf": [
        {
          "$ref": "#/$defs/FileCategory"
        },
        {
          "type": "null"
        }
      ],
      "default": null
    },
    "url": {
      "anyOf": [
        {
          "type": "string"
        },
        {
          "type": "null"
        }
      ],
      "default": null,
      "title": "Url"
    }
  },
  "required": [
    "name",
    "namespace",
    "contentType"
  ],
  "title": "SignURLPOSTFile",
  "type": "object"
}

Fields:

category pydantic-field 

category: FileCategory | None = None

content_type pydantic-field 

content_type: str

metadata pydantic-field 

metadata: list[dict[str, str]] | None = None

name pydantic-field 

name: str

namespace pydantic-field 

namespace: FileNamespace

url pydantic-field 

url: str | None = None