Lots

albert.collections.lots

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

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_by_id	Get a lot by its ID.
get_by_ids	Get a list of lots by their IDs.
list	Lists Lot entities with optional filters.
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_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"]]

list

list(
    *,
    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]

Lists 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 objects.

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

def list(
    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]:
    """
    Lists 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 objects.
    """
    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],
    )

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