Skip to content

Locations

albert.collections.locations

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

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

LocationCollection

LocationCollection(*, session: AlbertSession)

Bases: BaseCollection

LocationCollection is a collection class for managing Location entities in the Albert platform.

Parameters:

Name Type Description Default
session AlbertSession

The Albert session instance.

required

Methods:

Name Description
create

Creates a new Location entity.

delete

Deletes a Location entity.

get_by_id

Retrieves a location by its ID.

list

Searches for locations matching the provided criteria.

location_exists

Determines if a location, with the same name, exists in the collection.

update

Update a Location entity.

Source code in src/albert/collections/locations.py
def __init__(self, *, session: AlbertSession):
    """
    Initializes the LocationCollection with the provided session.

    Parameters
    ----------
    session : AlbertSession
        The Albert session instance.
    """
    super().__init__(session=session)
    self.base_path = f"/api/{LocationCollection._api_version}/locations"

base_path instance-attribute

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

create

create(*, location: Location) -> Location

Creates a new Location entity.

Parameters:

Name Type Description Default
location Location

The Location object to create.

required

Returns:

Type Description
Location

The created Location object.

Source code in src/albert/collections/locations.py
def create(self, *, location: Location) -> Location:
    """
    Creates a new Location entity.

    Parameters
    ----------
    location : Location
        The Location object to create.

    Returns
    -------
    Location
        The created Location object.
    """
    exists = self.location_exists(location=location)
    if exists:
        logging.warning(
            f"Location with name {location.name} matches an existing location. Returning the existing Location."
        )
        return exists

    payload = location.model_dump(by_alias=True, exclude_unset=True, mode="json")
    response = self.session.post(self.base_path, json=payload)

    return Location(**response.json())

delete

delete(*, id: str) -> None

Deletes a Location entity.

Parameters:

Name Type Description Default
id Str

The id of the Location object to delete.

required

Returns:

Type Description
None
Source code in src/albert/collections/locations.py
def delete(self, *, id: str) -> None:
    """
    Deletes a Location entity.

    Parameters
    ----------
    id : Str
        The id of the Location object to delete.

    Returns
    -------
    None
    """
    url = f"{self.base_path}/{id}"
    self.session.delete(url)

get_by_id

get_by_id(*, id: str) -> Location

Retrieves a location by its ID.

Parameters:

Name Type Description Default
id str

The ID of the location to retrieve.

required

Returns:

Type Description
Location

The Location object.

Source code in src/albert/collections/locations.py
def get_by_id(self, *, id: str) -> Location:
    """
    Retrieves a location by its ID.

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

    Returns
    -------
    Location
        The Location object.
    """
    url = f"{self.base_path}/{id}"
    response = self.session.get(url)
    return Location(**response.json())

list

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

Searches for locations matching the provided criteria.

Parameters:

Name Type Description Default
ids list[str] | None

The list of IDs to filter the locations, by default None. Max length is 100.

None
name str | list[str] | None

The name or names of locations to search for, by default None

None
country str | None

The country code of the country to filter the locations , by default None

None
exact_match bool

Whether to return exact matches only, by default False

False

Yields:

Type Description
Iterator[Location]

An iterator of Location objects matching the search criteria.

Source code in src/albert/collections/locations.py
def list(
    self,
    *,
    ids: list[str] | None = None,
    name: str | list[str] | None = None,
    country: str | None = None,
    exact_match: bool = False,
    limit: int = 50,
    start_key: str | None = None,
) -> Iterator[Location]:
    """Searches for locations matching the provided criteria.

    Parameters
    ----------
    ids: list[str] | None, optional
        The list of IDs to filter the locations, by default None.
        Max length is 100.
    name : str | list[str] | None, optional
        The name or names of locations to search for, by default None
    country : str | None, optional
        The country code of the country to filter the locations , by default None
    exact_match : bool, optional
        Whether to return exact matches only, by default False

    Yields
    ------
    Iterator[Location]
        An iterator of Location objects matching the search criteria.
    """
    params = {"limit": limit, "startKey": start_key, "country": country}
    if ids:
        params["id"] = ids
    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=lambda items: [Location(**item) for item in items],
    )

location_exists

location_exists(*, location: Location) -> Location | None

Determines if a location, with the same name, exists in the collection.

Parameters:

Name Type Description Default
location Location

The Location object to check

required

Returns:

Type Description
Location | None

The existing registered Location object if found, otherwise None.

Source code in src/albert/collections/locations.py
def location_exists(self, *, location: Location) -> Location | None:
    """Determines if a location, with the same name, exists in the collection.

    Parameters
    ----------
    location : Location
        The Location object to check

    Returns
    -------
    Location | None
        The existing registered Location object if found, otherwise None.
    """
    hits = self.list(name=location.name)
    if hits:
        for hit in hits:
            if hit and hit.name.lower() == location.name.lower():
                return hit
    return None

update

update(*, location: Location) -> Location

Update a Location entity.

Parameters:

Name Type Description Default
location Location

The Location object to update. The ID of the Location object must be provided.

required

Returns:

Type Description
Location

The updated Location object as returned by the server.

Source code in src/albert/collections/locations.py
def update(self, *, location: Location) -> Location:
    """Update a Location entity.

    Parameters
    ----------
    location : Location
        The Location object to update. The ID of the Location object must be provided.

    Returns
    -------
    Location
        The updated Location object as returned by the server.
    """
    # Fetch the current object state from the server or database
    current_object = self.get_by_id(id=location.id)
    # Generate the PATCH payload
    patch_payload = self._generate_patch_payload(
        existing=current_object,
        updated=location,
        stringify_values=True,
    )
    url = f"{self.base_path}/{location.id}"
    self.session.patch(url, json=patch_payload.model_dump(mode="json", by_alias=True))
    return self.get_by_id(id=location.id)