Storage Locations

albert.collections.storage_locations

AlbertHTTPError

AlbertHTTPError(response: Response)

Bases: AlbertException

Base class for all erors due to HTTP responses.

Source code in src/albert/exceptions.py

def __init__(self, response: requests.Response):
    message = self._format_message(response)
    super().__init__(message)
    self.response = response

response instance-attribute

response = response

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

EntityLink

Bases: BaseAlbertModel

Methods:

Name	Description
to_entity_link

id instance-attribute

id: str

name class-attribute instance-attribute

name: str | None = Field(default=None, exclude=True)

to_entity_link

to_entity_link() -> EntityLink

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

def to_entity_link(self) -> "EntityLink":
    # Convience method to return self, so you can call this method on objects that are already entity links
    return self

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

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_by_id	Get a storage location by its ID.
list	List storage locations with optional filtering.
update	Update a storage location.

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

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

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.list(
        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

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

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

list

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

List 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
Generator[StorageLocation, None, None]	description

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

def list(
    self,
    *,
    name: str | list[str] | None = None,
    exact_match: bool = False,
    location: str | Location | None = None,
    start_key: str | None = None,
    limit: int = 50,
) -> Generator[StorageLocation, None, None]:
    """List 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
    ------
    Generator[StorageLocation, None, None]
        _description_
    """

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

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

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)