Lots

albert.collections.lots

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

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: _ufor (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

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

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

Lot

Bases: BaseResource

A lot in Albert.

Attributes:

Name	Type	Description
id	LotId | None	The Albert ID of the lot. Set when the lot is retrieved from Albert.
inventory_id	InventoryId	The Albert ID of the inventory item associated with the lot.
task_id	str | None	The Albert ID of the task associated with the creation of lot. Optional.
notes	str | None	The notes associated with the lot. Optional.
expiration_date	str | None	The expiration date of the lot. YYYY-MM-DD format. Optional.
manufacturer_lot_number	str | None	The manufacturer lot number of the lot. Optional.
storage_location	StorageLocation | None	The storage location of the lot. Optional.
pack_size	str | None	The pack size of the lot. Optional. Used to calculate the cost per unit.
initial_quantity	NonNegativeFloat | None	The initial quantity of the lot. Optional.
cost	NonNegativeFloat | None	The cost of the lot. Optional.
inventory_on_hand	NonNegativeFloat	The inventory on hand of the lot.
owner	list[User] | None	The owners of the lot. Optional.
lot_number	str | None	The lot number of the lot. Optional.
external_barcode_id	str | None	The external barcode ID of the lot. Optional.
metadata	dict[str, str | list[EntityLink] | EntityLink] | None	The metadata of the lot. Optional. Metadata allowed values can be found using the Custom Fields API.
has_notes	bool	Whether the lot has notes. Read-only.
has_attachments	bool	Whether the lot has attachments. Read-only.
barcode_id	str	The barcode ID of the lot. Read-only.

Methods:

Name	Description
serialize_cost
serialize_initial_quantity
serialize_inventory_on_hand
validate_has_attachments
validate_has_notes

barcode_id class-attribute instance-attribute

barcode_id: str | None = Field(
    default=None, alias="barcodeId"
)

cost class-attribute instance-attribute

cost: NonNegativeFloat | None = Field(default=None)

expiration_date class-attribute instance-attribute

expiration_date: str | None = Field(
    None, alias="expirationDate"
)

external_barcode_id class-attribute instance-attribute

external_barcode_id: str | None = Field(
    None, alias="externalBarcodeId"
)

has_attachments class-attribute instance-attribute

has_attachments: bool | None = Field(
    default=None,
    alias="hasAttachments",
    exclude=True,
    frozen=True,
)

has_notes class-attribute instance-attribute

has_notes: bool | None = Field(
    default=None,
    alias="hasNotes",
    exclude=True,
    frozen=True,
)

id class-attribute instance-attribute

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

initial_quantity class-attribute instance-attribute

initial_quantity: NonNegativeFloat | None = Field(
    default=None, alias="initialQuantity"
)

inventory_id class-attribute instance-attribute

inventory_id: InventoryId = Field(alias='parentId')

inventory_on_hand class-attribute instance-attribute

inventory_on_hand: float = Field(alias='inventoryOnHand')

location class-attribute instance-attribute

location: SerializeAsEntityLink[Location] | None = Field(
    default=None,
    alias="Location",
    exclude=True,
    frozen=True,
)

lot_number class-attribute instance-attribute

lot_number: str | None = Field(None, alias='lotNumber')

manufacturer_lot_number class-attribute instance-attribute

manufacturer_lot_number: str | None = Field(
    None, alias="manufacturerLotNumber"
)

metadata class-attribute instance-attribute

metadata: dict[str, MetadataItem] | None = Field(
    alias="Metadata", default=None
)

notes class-attribute instance-attribute

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

owner class-attribute instance-attribute

owner: list[SerializeAsEntityLink[User]] | None = Field(
    default=None, alias="Owner"
)

pack_size class-attribute instance-attribute

pack_size: str | None = Field(None, alias='packSize')

parent_category class-attribute instance-attribute

parent_category: InventoryCategory | None = Field(
    default=None,
    alias="parentCategory",
    exclude=True,
    frozen=True,
)

parent_name class-attribute instance-attribute

parent_name: str | None = Field(
    default=None,
    alias="parentName",
    exclude=True,
    frozen=True,
)

parent_unit class-attribute instance-attribute

parent_unit: str | None = Field(
    default=None,
    alias="parentUnit",
    exclude=True,
    frozen=True,
)

status class-attribute instance-attribute

status: LotStatus | None = Field(default=None)

storage_location class-attribute instance-attribute

storage_location: (
    SerializeAsEntityLink[StorageLocation] | None
) = Field(alias="StorageLocation", default=None)

task_id class-attribute instance-attribute

task_id: str | None = Field(default=None, alias='taskId')

serialize_cost

serialize_cost(cost: NonNegativeFloat)

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

@field_serializer("cost", return_type=str)
def serialize_cost(self, cost: NonNegativeFloat):
    return str(cost)

serialize_initial_quantity

serialize_initial_quantity(
    initial_quantity: NonNegativeFloat,
)

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

@field_serializer("initial_quantity", return_type=str)
def serialize_initial_quantity(self, initial_quantity: NonNegativeFloat):
    return str(initial_quantity)

serialize_inventory_on_hand

serialize_inventory_on_hand(
    inventory_on_hand: NonNegativeFloat,
)

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

@field_serializer("inventory_on_hand", return_type=str)
def serialize_inventory_on_hand(self, inventory_on_hand: NonNegativeFloat):
    return str(inventory_on_hand)

validate_has_attachments

validate_has_attachments(value: Any) -> Any

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

@field_validator("has_attachments", mode="before")
def validate_has_attachments(cls, value: Any) -> Any:
    if value == "1":
        return True
    elif value == "0":
        return False
    return value

validate_has_notes

validate_has_notes(value: Any) -> Any

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

@field_validator("has_notes", mode="before")
def validate_has_notes(cls, value: Any) -> Any:
    if value == "1":
        return True
    elif value == "0":
        return False
    return value

LotCollection

LotCollection(*, session: AlbertSession)

Bases: BaseCollection

LotCollection is a collection class for managing Lot entities in the Albert platform.

Parameters:

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

Methods:

Name	Description
create
delete	Delete a lot by its ID.
get_all	Get all Lot entities with optional filters.
get_by_id	Get a lot by its ID.
get_by_ids	Get a list of lots by their IDs.
update	Update a lot.

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

def __init__(self, *, session: AlbertSession):
    """A collection for interacting with Lots in Albert.

    Parameters
    ----------
    session : AlbertSession
        An Albert session instance.
    """
    super().__init__(session=session)
    self.base_path = f"/api/{LotCollection._api_version}/lots"

base_path instance-attribute

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

create

create(*, lots: list[Lot]) -> list[Lot]

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

def create(self, *, lots: list[Lot]) -> list[Lot]:
    # TODO: Once thi endpoint is fixed, go back to passing the whole list at once
    payload = [lot.model_dump(by_alias=True, exclude_none=True, mode="json") for lot in lots]
    all_lots = []
    for lot in payload:
        response = self.session.post(self.base_path, json=[lot])
        all_lots.append(Lot(**response.json()[0]))
    # response = self.session.post(self.base_path, json=payload)
    # return [Lot(**lot) for lot in response.json().get("CreatedLots", [])]
    return all_lots

delete

delete(*, id: str) -> None

Delete a lot by its ID.

Parameters:

Name	Type	Description	Default
id	str	The ID of the lot to delete.	required

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

def delete(self, *, id: str) -> None:
    """Delete a lot by its ID.

    Parameters
    ----------
    id : str
        The ID of the lot to delete.
    """
    url = f"{self.base_path}?id={id}"
    self.session.delete(url)

get_all

get_all(
    *,
    limit: int = 100,
    start_key: str | None = None,
    parent_id: str | None = None,
    inventory_id: str | None = None,
    barcode_id: str | None = None,
    parent_id_category: str | None = None,
    inventory_on_hand: str | None = None,
    location_id: str | None = None,
    exact_match: bool = False,
    begins_with: bool = False,
) -> Iterator[Lot]

Get all Lot entities with optional filters.

Parameters:

Name	Type	Description	Default
limit	int	The maximum number of Lots to return, by default 100.	100
start_key	Optional[str]	The primary key of the first item to evaluate for pagination.	None
parent_id	Optional[str]	Fetches list of lots for a parentId (inventory).	None
inventory_id	Optional[str]	Fetches list of lots for an inventory.	None
barcode_id	Optional[str]	Fetches list of lots for a barcodeId.	None
parent_id_category	Optional[str]	Fetches list of lots for a parentIdCategory (e.g., RawMaterials, Consumables).	None
inventory_on_hand	Optional[str]	Fetches records based on inventoryOnHand (lteZero, gtZero, eqZero).	None
location_id	Optional[str]	Fetches list of lots for a locationId.	None
exact_match	bool	Determines if barcodeId field should be an exact match, by default False.	False
begins_with	bool	Determines if barcodeId begins with a certain value, by default False.	False

Yields:

Type	Description
Iterator[Lot]	An iterator of Lot entities.

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

def get_all(
    self,
    *,
    limit: int = 100,
    start_key: str | None = None,
    parent_id: str | None = None,
    inventory_id: str | None = None,
    barcode_id: str | None = None,
    parent_id_category: str | None = None,
    inventory_on_hand: str | None = None,
    location_id: str | None = None,
    exact_match: bool = False,
    begins_with: bool = False,
) -> Iterator[Lot]:
    """
    Get all Lot entities with optional filters.

    Parameters
    ----------
    limit : int, optional
        The maximum number of Lots to return, by default 100.
    start_key : Optional[str], optional
        The primary key of the first item to evaluate for pagination.
    parent_id : Optional[str], optional
        Fetches list of lots for a parentId (inventory).
    inventory_id : Optional[str], optional
        Fetches list of lots for an inventory.
    barcode_id : Optional[str], optional
        Fetches list of lots for a barcodeId.
    parent_id_category : Optional[str], optional
        Fetches list of lots for a parentIdCategory (e.g., RawMaterials, Consumables).
    inventory_on_hand : Optional[str], optional
        Fetches records based on inventoryOnHand (lteZero, gtZero, eqZero).
    location_id : Optional[str], optional
        Fetches list of lots for a locationId.
    exact_match : bool, optional
        Determines if barcodeId field should be an exact match, by default False.
    begins_with : bool, optional
        Determines if barcodeId begins with a certain value, by default False.

    Yields
    -------
    Iterator[Lot]
        An iterator of Lot entities.
    """
    params = {
        "limit": limit,
        "startKey": start_key,
        "parentId": parent_id,
        "inventoryId": inventory_id,
        "barcodeId": barcode_id,
        "parentIdCategory": parent_id_category,
        "inventoryOnHand": inventory_on_hand,
        "locationId": location_id,
        "exactMatch": json.dumps(exact_match),
        "beginsWith": json.dumps(begins_with),
    }
    return AlbertPaginator(
        mode=PaginationMode.KEY,
        path=self.base_path,
        session=self.session,
        params=params,
        deserialize=lambda items: [Lot(**item) for item in items],
    )

get_by_id

get_by_id(*, id: str) -> Lot

Get a lot by its ID.

Parameters:

Name	Type	Description	Default
id	str	The ID of the lot to get.	required

Returns:

Type	Description
Lot	The lot with the provided ID.

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

def get_by_id(self, *, id: str) -> Lot:
    """Get a lot by its ID.

    Parameters
    ----------
    id : str
        The ID of the lot to get.

    Returns
    -------
    Lot
        The lot with the provided ID.
    """
    url = f"{self.base_path}/{id}"
    response = self.session.get(url)
    return Lot(**response.json())

get_by_ids

get_by_ids(*, ids: list[str]) -> list[Lot]

Get a list of lots by their IDs.

Parameters:

Name	Type	Description	Default
ids	list[str]	A list of lot IDs to get.	required

Returns:

Type	Description
list[Lot]	A list of lots with the provided IDs.

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

def get_by_ids(self, *, ids: list[str]) -> list[Lot]:
    """Get a list of lots by their IDs.

    Parameters
    ----------
    ids : list[str]
        A list of lot IDs to get.

    Returns
    -------
    list[Lot]
        A list of lots with the provided IDs.
    """
    url = f"{self.base_path}/ids"
    response = self.session.get(url, params={"id": ids})
    return [Lot(**lot) for lot in response.json()["Items"]]

update

update(*, lot: Lot) -> Lot

Update a lot.

Parameters:

Name	Type	Description	Default
lot	Lot	The updated lot object.	required

Returns:

Type	Description
Lot	The updated Lot entity as returned by the server.

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

def update(self, *, lot: Lot) -> Lot:
    """Update a lot.

    Parameters
    ----------
    lot : Lot
        The updated lot object.

    Returns
    -------
    Lot
        The updated Lot entity as returned by the server.
    """
    existing_lot = self.get_by_id(id=lot.id)
    patch_data = self._generate_patch_payload(existing=existing_lot, updated=lot)
    url = f"{self.base_path}/{lot.id}"

    self.session.patch(url, json=patch_data.model_dump(mode="json", by_alias=True))

    return self.get_by_id(id=lot.id)

PaginationMode

Bases: str, Enum

KEY class-attribute instance-attribute

KEY = 'key'

OFFSET class-attribute instance-attribute

OFFSET = 'offset'