Skip to content

Locations

albert.collections.locations

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

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.
exists

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

Get all Location entities matching the provided criteria.
get_by_id

Retrieves a location by its ID.
update

Update a Location entity.
Source code in src/albert/collections/locations.py 
17
18
19
20
21
22
23
24
25
26
27
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 entity to create.

 required

Returns:

Type Description
Location

The created Location entity.
Source code in src/albert/collections/locations.py 
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
def create(self, *, location: Location) -> Location:
    """
    Creates a new Location entity.

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

    Returns
    -------
    Location
        The created Location entity.
    """
    exists = self.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 entity to delete.

 required

Returns:

Type Description
None
Source code in src/albert/collections/locations.py 
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
def delete(self, *, id: str) -> None:
    """
    Deletes a Location entity.

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

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

exists 

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 entity to check

 required

Returns:

Type Description
Location | None

The existing registered Location entity if found, otherwise None.
Source code in src/albert/collections/locations.py 
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
def exists(self, *, location: Location) -> Location | None:
    """Determines if a location, with the same name, exists in the collection.

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

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

get_all 

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

Get all Location entities 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 entities matching the search criteria.
Source code in src/albert/collections/locations.py 
29
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
67
68
69
70
def get_all(
    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]:
    """Get all Location entities 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 entities 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],
    )

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 
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
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())

update 

update(*, location: Location) -> Location

Update a Location entity.

Parameters:

Name Type Description Default
location Location

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

 required

Returns:

Type Description
Location

The updated Location entity as returned by the server.
Source code in src/albert/collections/locations.py 
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
def update(self, *, location: Location) -> Location:
    """Update a Location entity.

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

    Returns
    -------
    Location
        The updated Location entity 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)

PaginationMode

Bases: str, Enum

KEY class-attribute instance-attribute 

KEY = 'key'

OFFSET class-attribute instance-attribute 

OFFSET = 'offset'