Skip to content

Files

albert.collections.files

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 
27
28
def __init__(self, *, session: AlbertSession):
    self.session = session

session 

session = session

FileCategory

Bases: str, Enum

SDS 

SDS = 'SDS'

OTHER 

OTHER = 'Other'

FileInfo

Bases: BaseAlbertModel

name 

name: str

size 

size: int

etag 

etag: str

namespace 

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

content_type 

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

last_modified 

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

metadata 

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

FileNamespace

Bases: str, Enum

AGENT 

AGENT = 'agent'

BREAKTHROUGH 

BREAKTHROUGH = 'breakthrough'

PIPELINE 

PIPELINE = 'pipeline'

PUBLIC 

PUBLIC = 'public'

RESULT 

RESULT = 'result'

SDS 

SDS = 'sds'

SignURLPOST

Bases: BaseAlbertModel

files 

files: list[SignURLPOSTFile]

SignURLPOSTFile

Bases: BaseAlbertModel

name 

name: str

namespace 

namespace: FileNamespace

content_type 

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

metadata 

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

category 

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

url 

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

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
Source code in src/albert/session.py 
28
29
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,
    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 

base_url = base_url

request 

request(
    method: str, path: str, *args, **kwargs
) -> Response
Source code in src/albert/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

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

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 
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
        _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 
 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
        _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 
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
        _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 
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})