Skip to content

Storage Locations

albert.collections.storage_locations

logger module-attribute 

logger = create_logger()

AlbertHTTPError 

AlbertHTTPError(response: Response)

Bases: AlbertException

Base class for all erors due to HTTP responses.

Source code in src/albert/exceptions.py 
18
19
20
21
def __init__(self, response: requests.Response):
    message = self._format_message(response)
    super().__init__(message)
    self.response = response

response instance-attribute 

response = response

AlbertPaginator 

AlbertPaginator(
    *,
    path: str,
    mode: PaginationMode,
    session: AlbertSession,
    deserialize: Callable[
        [Iterable[dict]], Iterable[ItemType]
    ],
    params: dict[str, str] | None = None,
)

Bases: Iterator[ItemType]

Helper class for pagination through Albert endpoints.

Two pagination modes are possible: - Offset-based via by the offset query parameter - Key-based via by the startKey query parameter and 'lastKey' response field

A custom deserialize function is provided when additional logic is required to load the raw items returned by the search listing, e.g., making additional Albert API calls.

Source code in src/albert/core/pagination.py 
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
def __init__(
    self,
    *,
    path: str,
    mode: PaginationMode,
    session: AlbertSession,
    deserialize: Callable[[Iterable[dict]], Iterable[ItemType]],
    params: dict[str, str] | None = None,
):
    self.path = path
    self.mode = mode
    self.session = session
    self.deserialize = deserialize

    params = params or {}
    self.params = {k: v for k, v in params.items() if v is not None}

    if "startKey" in self.params:
        self.params["startKey"] = quote_plus(self.params["startKey"])

    self._iterator = self._create_iterator()

deserialize instance-attribute 

deserialize = deserialize

mode instance-attribute 

mode = mode

params instance-attribute 

params = {k: _cfor (k, v) in items() if v is not None}

path instance-attribute 

path = path

session instance-attribute 

session = session

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

Location

Bases: BaseResource

A location in Albert.

Attributes:

Name Type Description
name str

The name of the location.
id str | None

The Albert ID of the location. Set when the location is retrieved from Albert.
latitude float

The latitude of the location.
longitude float

The longitude of the location.
address str

The address of the location.
country str | None

The country code of the location. Must be two characters long.

address instance-attribute 

address: str

country class-attribute instance-attribute 

country: str | None = Field(
    None, max_length=2, min_length=2
)

id class-attribute instance-attribute 

id: str | None = Field(None, alias='albertId')

latitude class-attribute instance-attribute 

latitude: float = Field()

longitude class-attribute instance-attribute 

longitude: float = Field()

name instance-attribute 

name: str

PaginationMode

Bases: str, Enum

KEY class-attribute instance-attribute 

KEY = 'key'

OFFSET class-attribute instance-attribute 

OFFSET = 'offset'

StorageLocation

Bases: BaseResource

A storage location entity. For example, a specific flammables cabinet or a storage room.

Attributes:

Name Type Description
name str

The name of the storage location.
id str | None

The Albert ID of the storage location. Set when the storage location is retrieved from Albert.
location Location

The location entity link of the storage location.

id class-attribute instance-attribute 

id: str | None = Field(alias='albertId', default=None)

location class-attribute instance-attribute 

location: SerializeAsEntityLink[Location] = Field(
    alias="Location"
)

name class-attribute instance-attribute 

name: str = Field(
    alias="name", min_length=2, max_length=255
)

StorageLocationsCollection 

StorageLocationsCollection(*, session: AlbertSession)

Bases: BaseCollection

StorageLocationsCollection is a collection class for managing StorageLoction entities in the Albert platform.

Parameters:

Name Type Description Default
session AlbertSession

The Albert Session information

 required

Methods:

Name Description
create

Create a new storage location.
delete

Delete a storage location by its ID.
get_all

Get all storage locations with optional filtering.
get_by_id

Get a storage location by its ID.
update

Update a storage location.
Source code in src/albert/collections/storage_locations.py 
21
22
23
24
25
26
27
28
29
30
def __init__(self, *, session: AlbertSession):
    """Initialize the StorageLocationsCollection.

    Parameters
    ----------
    session : AlbertSession
        The Albert Session information
    """
    super().__init__(session=session)
    self.base_path = f"/api/{StorageLocationsCollection._api_version}/storagelocations"

base_path instance-attribute 

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

create 

create(
    *, storage_location: StorageLocation
) -> StorageLocation

Create a new storage location.

Parameters:

Name Type Description Default
storage_location StorageLocation

The storage location to create.

 required

Returns:

Type Description
StorageLocation

The created storage location.
Source code in src/albert/collections/storage_locations.py 
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
def create(self, *, storage_location: StorageLocation) -> StorageLocation:
    """Create a new storage location.

    Parameters
    ----------
    storage_location : StorageLocation
        The storage location to create.

    Returns
    -------
    StorageLocation
        The created storage location.
    """
    matching = self.get_all(
        name=storage_location.name, location=storage_location.location, exact_match=True
    )
    for m in matching:
        if m.name.lower() == storage_location.name.lower():
            logging.warning(
                f"Storage location with name {storage_location.name} already exists, returning existing."
            )
            return m

    path = self.base_path
    response = self.session.post(
        path, json=storage_location.model_dump(by_alias=True, exclude_none=True, mode="json")
    )
    return StorageLocation(**response.json())

delete 

delete(*, id: str) -> None

Delete a storage location by its ID.

Parameters:

Name Type Description Default
id str

The ID of the storage location to delete.

 required
Source code in src/albert/collections/storage_locations.py 
130
131
132
133
134
135
136
137
138
139
def delete(self, *, id: str) -> None:
    """Delete a storage location by its ID.

    Parameters
    ----------
    id : str
        The ID of the storage location to delete.
    """
    path = f"{self.base_path}/{id}"
    self.session.delete(path)

get_all 

get_all(
    *,
    name: str | list[str] | None = None,
    exact_match: bool = False,
    location: str | Location | None = None,
    start_key: str | None = None,
    limit: int = 50,
) -> Iterator[StorageLocation]

Get all storage locations with optional filtering.

Parameters:

Name Type Description Default
name str | list[str] | None

The name or names of the storage locations to filter by, by default None

 None
exact_match bool

Whether to perform an exact match on the name, by default False

 False
location str | Location | None

The location ID or Location object to filter by, by default None

 None

Yields:

Type Description
Iterator[StorageLocation]

An iterator of StorageLocation items matching the search criteria.
Source code in src/albert/collections/storage_locations.py 
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
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
def get_all(
    self,
    *,
    name: str | list[str] | None = None,
    exact_match: bool = False,
    location: str | Location | None = None,
    start_key: str | None = None,
    limit: int = 50,
) -> Iterator[StorageLocation]:
    """Get all storage locations with optional filtering.

    Parameters
    ----------
    name : str | list[str] | None, optional
        The name or names of the storage locations to filter by, by default None
    exact_match : bool, optional
        Whether to perform an exact match on the name, by default False
    location : str | Location | None, optional
        The location ID or Location object to filter by, by default None

    Yields
    ------
    Iterator[StorageLocation]
        An iterator of StorageLocation items matching the search criteria.
    """

    # Remove explicit hydration when SUP-410 is fixed
    def deserialize(items: list[dict]) -> Iterator[StorageLocation]:
        for x in items:
            id = x["albertId"]
            try:
                yield self.get_by_id(id=id)
            except AlbertHTTPError as e:
                logger.warning(f"Error fetching storage location {id}: {e}")

    params = {
        "limit": limit,
        "locationId": location.id if isinstance(location, Location | EntityLink) else location,
        "startKey": start_key,
    }
    if name:
        params["name"] = [name] if isinstance(name, str) else name
        params["exactMatch"] = json.dumps(exact_match)

    return AlbertPaginator(
        mode=PaginationMode.KEY,
        path=self.base_path,
        session=self.session,
        params=params,
        deserialize=deserialize,
    )

get_by_id 

get_by_id(*, id: str) -> StorageLocation

Get a storage location by its ID.

Parameters:

Name Type Description Default
id str

The ID of the storage location to retrieve.

 required

Returns:

Type Description
StorageLocation

The retrieved storage location with the given ID.
Source code in src/albert/collections/storage_locations.py 
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
def get_by_id(self, *, id: str) -> StorageLocation:
    """Get a storage location by its ID.

    Parameters
    ----------
    id : str
        The ID of the storage location to retrieve.

    Returns
    -------
    StorageLocation
        The retrieved storage location with the given ID.
    """
    path = f"{self.base_path}/{id}"
    response = self.session.get(path)
    return StorageLocation(**response.json())

update 

update(
    *, storage_location: StorageLocation
) -> StorageLocation

Update a storage location.

Parameters:

Name Type Description Default
storage_location StorageLocation

The storage location to update.

 required

Returns:

Type Description
StorageLocation

The updated storage location as returned by the server.
Source code in src/albert/collections/storage_locations.py 
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
def update(self, *, storage_location: StorageLocation) -> StorageLocation:
    """Update a storage location.

    Parameters
    ----------
    storage_location : StorageLocation
        The storage location to update.

    Returns
    -------
    StorageLocation
        The updated storage location as returned by the server.
    """
    path = f"{self.base_path}/{storage_location.id}"
    payload = self._generate_patch_payload(
        existing=self.get_by_id(id=storage_location.id),
        updated=storage_location,
    )
    self.session.patch(path, json=payload.model_dump(mode="json", by_alias=True))
    return self.get_by_id(id=storage_location.id)