Skip to content

Users

albert.collections.users

logger module-attribute 

logger = create_logger()

AlbertHTTPError 

AlbertHTTPError(response: Response)

Bases: AlbertException

Base class for all erors due to HTTP responses.

Source code in src/albert/exceptions.py 
18
19
20
21
def __init__(self, response: requests.Response):
    message = self._format_message(response)
    super().__init__(message)
    self.response = response

response instance-attribute 

response = response

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

PaginationMode

Bases: str, Enum

KEY class-attribute instance-attribute 

KEY = 'key'

OFFSET class-attribute instance-attribute 

OFFSET = 'offset'

Status

Bases: str, Enum

The status of a resource

ACTIVE class-attribute instance-attribute 

ACTIVE = 'active'

INACTIVE class-attribute instance-attribute 

INACTIVE = 'inactive'

User

Bases: BaseResource

Represents a User on the Albert Platform

Attributes:

Name Type Description
name str

The name of the user.
id str | None

The Albert ID of the user. Set when the user is retrieved from Albert.
location Location | None

The location of the user.
email EmailStr | None

The email of the user.
roles list[Role]

The roles of the user.
user_class UserClass

The ACL class level of the user.
metadata dict[str, str | list[EntityLink] | EntityLink] | None

Methods:

Name Description
to_note_mention

Convert the user to a note mention string.

email class-attribute instance-attribute 

email: EmailStr = Field(default=None, alias='email')

id class-attribute instance-attribute 

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

location class-attribute instance-attribute 

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

metadata class-attribute instance-attribute 

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

name instance-attribute 

name: str

roles class-attribute instance-attribute 

roles: list[SerializeAsEntityLink[Role]] = Field(
    max_length=1, default_factory=list, alias="Roles"
)

user_class class-attribute instance-attribute 

user_class: UserClass = Field(
    default=STANDARD, alias="userClass"
)

to_note_mention 

to_note_mention() -> str

Convert the user to a note mention string.

Returns:

Type Description
str

The note mention string.
Source code in src/albert/resources/users.py 
60
61
62
63
64
65
66
67
68
def to_note_mention(self) -> str:
    """Convert the user to a note mention string.

    Returns
    -------
    str
        The note mention string.
    """
    return f"@{self.name}#{self.id}#"

UserCollection 

UserCollection(*, session: AlbertSession)

Bases: BaseCollection

UserCollection is a collection class for managing User entities in the Albert platform.

Parameters:

Name Type Description Default
session AlbertSession

The Albert session instance.

 required

Methods:

Name Description
create

Create a new User
get_all

Retrieve fully hydrated User entities with optional filters.
get_by_id

Retrieves a User by its ID.
get_current_user

Retrieves the current authenticated user.
search

Searches for Users matching the provided criteria.
update

Update a User entity.
Source code in src/albert/collections/users.py 
20
21
22
23
24
25
26
27
28
29
30
def __init__(self, *, session: AlbertSession):
    """
    Initializes the UserCollection with the provided session.

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

base_path instance-attribute 

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

create 

create(*, user: User) -> User

Create a new User

Parameters:

Name Type Description Default
user User

The user to create

 required

Returns:

Type Description
User

The created User
Source code in src/albert/collections/users.py 
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
def create(self, *, user: User) -> User:  # pragma: no cover
    """Create a new User

    Parameters
    ----------
    user : User
        The user to create

    Returns
    -------
    User
        The created User
    """

    response = self.session.post(
        self.base_path,
        json=user.model_dump(by_alias=True, exclude_none=True, mode="json"),
    )
    return User(**response.json())

get_all 

get_all(
    *,
    limit: int = 100,
    status: Status | None = None,
    type: UserFilterType | None = None,
    id: list[str] | None = None,
    start_key: str | None = None,
) -> Iterator[User]

Retrieve fully hydrated User entities with optional filters.

This method uses get_by_id to hydrate the results for convenience. Use :meth:search for better performance.

Parameters:

Name Type Description Default
limit int

Max results per page.

 100
status Status

Filter by user status.

 None
type UserFilterType

Attribute name to filter by (e.g., 'role').

 None
id list[str]

Values of the attribute to filter on.

 None
start_key Optional[str]

The starting point for the next set of results, by default None.

 None

Returns:

Type Description
Iterator[User]

Fully hydrated User entities.
Source code in src/albert/collections/users.py 
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
def get_all(
    self,
    *,
    limit: int = 100,
    status: Status | None = None,
    type: UserFilterType | None = None,
    id: list[str] | None = None,
    start_key: str | None = None,
) -> Iterator[User]:
    """
    Retrieve fully hydrated User entities with optional filters.

    This method uses `get_by_id` to hydrate the results for convenience.
    Use :meth:`search` for better performance.

    Parameters
    ----------
    limit : int
        Max results per page.
    status : Status, optional
        Filter by user status.
    type : UserFilterType
        Attribute name to filter by (e.g., 'role').
    id : list[str], optional
        Values of the attribute to filter on.
    start_key : Optional[str], optional
        The starting point for the next set of results, by default None.

    Returns
    -------
    Iterator[User]
        Fully hydrated User entities.
    """
    params = {
        "limit": limit,
        "status": status,
        "type": type.value if type else None,
        "id": id,
        "startKey": start_key,
    }

    def deserialize(items: list[dict]) -> Iterator[User]:
        for item in items:
            user_id = item.get("albertId")
            if user_id:
                try:
                    yield self.get_by_id(id=user_id)
                except AlbertHTTPError as e:
                    logger.warning(f"Error fetching user '{user_id}': {e}")

    return AlbertPaginator(
        mode=PaginationMode.KEY,
        path=self.base_path,
        session=self.session,
        params=params,
        deserialize=deserialize,
    )

get_by_id 

get_by_id(*, id: str) -> User

Retrieves a User by its ID.

Parameters:

Name Type Description Default
id str

The ID of the user to retrieve.

 required

Returns:

Type Description
User

The User object.
Source code in src/albert/collections/users.py 
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
def get_by_id(self, *, id: str) -> User:
    """
    Retrieves a User by its ID.

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

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

get_current_user 

get_current_user() -> User

Retrieves the current authenticated user.

Returns:

Type Description
User

The current User object.
Source code in src/albert/collections/users.py 
32
33
34
35
36
37
38
39
40
41
42
def get_current_user(self) -> User:
    """
    Retrieves the current authenticated user.

    Returns
    -------
    User
        The current User object.
    """
    claims = jwt.decode(self.session._access_token, options={"verify_signature": False})
    return self.get_by_id(id=claims["id"])

search 

search(
    *, params: UserFilterParams | None = None
) -> Iterator[UserSearchItem]

Searches for Users matching the provided criteria.

⚠️ This method returns partial (unhydrated) search results for performance. To retrieve fully detailed objects, use :meth:get_all instead.

Parameters:

Name Type Description Default
params UserFilterParams

Structured search filters for user listing.

 None

Returns:

Type Description
Iterator[User]

An iterator of partial User entities.
Source code in src/albert/collections/users.py 
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
def search(self, *, params: UserFilterParams | None = None) -> Iterator[UserSearchItem]:
    """
    Searches for Users matching the provided criteria.

    ⚠️ This method returns partial (unhydrated) search results for performance.
    To retrieve fully detailed objects, use :meth:`get_all` instead.

    Parameters
    ----------
    params : UserFilterParams, optional
        Structured search filters for user listing.

    Returns
    -------
    Iterator[User]
        An iterator of partial User entities.
    """

    params = params or UserFilterParams()
    query_params = params.model_dump(exclude_none=True, by_alias=True)

    return AlbertPaginator(
        mode=PaginationMode.OFFSET,
        path=f"{self.base_path}/search",
        session=self.session,
        params=query_params,
        deserialize=lambda items: [
            UserSearchItem(**item)._bind_collection(self) for item in items
        ],
    )

update 

update(*, user: User) -> User

Update a User entity.

Parameters:

Name Type Description Default
user User

The updated User entity.

 required

Returns:

Type Description
User

The updated User entity as returned by the server.
Source code in src/albert/collections/users.py 
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
def update(self, *, user: User) -> User:
    """Update a User entity.

    Parameters
    ----------
    user : User
        The updated User entity.

    Returns
    -------
    User
        The updated User entity as returned by the server.
    """
    # Fetch the current object state from the server or database
    current_object = self.get_by_id(id=user.id)

    # Generate the PATCH payload
    payload = self._generate_patch_payload(existing=current_object, updated=user)

    url = f"{self.base_path}/{user.id}"
    self.session.patch(url, json=payload.model_dump(mode="json", by_alias=True))

    updated_user = self.get_by_id(id=user.id)
    return updated_user

UserFilterParams pydantic-model

Bases: BaseAlbertModel

Structured filters for searching users, with API-compatible query parameter names.

Show JSON schema: 
{
  "$defs": {
    "OrderBy": {
      "enum": [
        "desc",
        "asc"
      ],
      "title": "OrderBy",
      "type": "string"
    },
    "Status": {
      "description": "The status of a resource",
      "enum": [
        "active",
        "inactive"
      ],
      "title": "Status",
      "type": "string"
    }
  },
  "description": "Structured filters for searching users, with API-compatible query parameter names.",
  "properties": {
    "limit": {
      "default": 50,
      "maximum": 1000,
      "minimum": 1,
      "title": "Limit",
      "type": "integer"
    },
    "offset": {
      "anyOf": [
        {
          "type": "integer"
        },
        {
          "type": "null"
        }
      ],
      "default": null,
      "title": "Offset"
    },
    "text": {
      "anyOf": [
        {
          "type": "string"
        },
        {
          "type": "null"
        }
      ],
      "default": null,
      "title": "Text"
    },
    "sortBy": {
      "anyOf": [
        {
          "type": "string"
        },
        {
          "type": "null"
        }
      ],
      "default": null,
      "title": "Sortby"
    },
    "order": {
      "$ref": "#/$defs/OrderBy",
      "default": "desc"
    },
    "roles": {
      "anyOf": [
        {
          "items": {
            "type": "string"
          },
          "type": "array"
        },
        {
          "type": "null"
        }
      ],
      "default": null,
      "title": "Roles"
    },
    "teams": {
      "anyOf": [
        {
          "items": {
            "type": "string"
          },
          "type": "array"
        },
        {
          "type": "null"
        }
      ],
      "default": null,
      "title": "Teams"
    },
    "locations": {
      "anyOf": [
        {
          "items": {
            "type": "string"
          },
          "type": "array"
        },
        {
          "type": "null"
        }
      ],
      "default": null,
      "title": "Locations"
    },
    "status": {
      "anyOf": [
        {
          "items": {
            "$ref": "#/$defs/Status"
          },
          "type": "array"
        },
        {
          "type": "null"
        }
      ],
      "default": null,
      "title": "Status"
    },
    "userId": {
      "anyOf": [
        {
          "items": {
            "type": "string"
          },
          "type": "array"
        },
        {
          "type": "null"
        }
      ],
      "default": null,
      "title": "Userid"
    },
    "subscription": {
      "anyOf": [
        {
          "items": {
            "type": "string"
          },
          "type": "array"
        },
        {
          "type": "null"
        }
      ],
      "default": null,
      "title": "Subscription"
    },
    "searchFields": {
      "anyOf": [
        {
          "items": {
            "type": "string"
          },
          "type": "array"
        },
        {
          "type": "null"
        }
      ],
      "default": null,
      "title": "Searchfields"
    },
    "facetText": {
      "anyOf": [
        {
          "type": "string"
        },
        {
          "type": "null"
        }
      ],
      "default": null,
      "title": "Facettext"
    },
    "facetField": {
      "anyOf": [
        {
          "type": "string"
        },
        {
          "type": "null"
        }
      ],
      "default": null,
      "title": "Facetfield"
    },
    "containsField": {
      "anyOf": [
        {
          "items": {
            "type": "string"
          },
          "type": "array"
        },
        {
          "type": "null"
        }
      ],
      "default": null,
      "title": "Containsfield"
    },
    "containsText": {
      "anyOf": [
        {
          "items": {
            "type": "string"
          },
          "type": "array"
        },
        {
          "type": "null"
        }
      ],
      "default": null,
      "title": "Containstext"
    },
    "mentions": {
      "anyOf": [
        {
          "type": "boolean"
        },
        {
          "type": "null"
        }
      ],
      "default": null,
      "title": "Mentions"
    }
  },
  "title": "UserFilterParams",
  "type": "object"
}

Fields:

contains_field pydantic-field 

contains_field: list[str] | None = None

contains_text pydantic-field 

contains_text: list[str] | None = None

facet_field pydantic-field 

facet_field: str | None = None

facet_text pydantic-field 

facet_text: str | None = None

limit pydantic-field 

limit: int = 50

locations pydantic-field 

locations: list[str] | None = None

mentions pydantic-field 

mentions: bool | None = None

offset pydantic-field 

offset: int | None = None

order_by pydantic-field 

order_by: OrderBy = DESCENDING

roles pydantic-field 

roles: list[str] | None = None

search_fields pydantic-field 

search_fields: list[str] | None = None

sort_by pydantic-field 

sort_by: str | None = None

status pydantic-field 

status: list[Status] | None = None

subscription pydantic-field 

subscription: list[str] | None = None

teams pydantic-field 

teams: list[str] | None = None

text pydantic-field 

text: str | None = None

user_id pydantic-field 

user_id: list[str] | None = None

UserFilterType

Bases: str, Enum

ROLE class-attribute instance-attribute 

ROLE = 'role'

UserSearchItem pydantic-model

Bases: BaseAlbertModel, HydrationMixin[User]

Partial user entity as returned by the search.

Show JSON schema: 
{
  "$defs": {
    "UserClass": {
      "description": "The ACL class level of the user",
      "enum": [
        "guest",
        "standard",
        "trusted",
        "privileged",
        "admin"
      ],
      "title": "UserClass",
      "type": "string"
    },
    "UserSearchRoleItem": {
      "properties": {
        "roleId": {
          "title": "Roleid",
          "type": "string"
        },
        "roleName": {
          "title": "Rolename",
          "type": "string"
        }
      },
      "required": [
        "roleId",
        "roleName"
      ],
      "title": "UserSearchRoleItem",
      "type": "object"
    }
  },
  "description": "Partial user entity as returned by the search.",
  "properties": {
    "name": {
      "title": "Name",
      "type": "string"
    },
    "albertId": {
      "anyOf": [
        {
          "type": "string"
        },
        {
          "type": "null"
        }
      ],
      "default": null,
      "title": "Albertid"
    },
    "email": {
      "anyOf": [
        {
          "format": "email",
          "type": "string"
        },
        {
          "type": "null"
        }
      ],
      "default": null,
      "title": "Email"
    },
    "userClass": {
      "$ref": "#/$defs/UserClass",
      "default": "standard"
    },
    "lastLoginTime": {
      "anyOf": [
        {
          "format": "date-time",
          "type": "string"
        },
        {
          "type": "null"
        }
      ],
      "default": null,
      "title": "Lastlogintime"
    },
    "location": {
      "anyOf": [
        {
          "type": "string"
        },
        {
          "type": "null"
        }
      ],
      "default": null,
      "title": "Location"
    },
    "locationId": {
      "anyOf": [
        {
          "type": "string"
        },
        {
          "type": "null"
        }
      ],
      "default": null,
      "title": "Locationid"
    },
    "role": {
      "items": {
        "$ref": "#/$defs/UserSearchRoleItem"
      },
      "maxItems": 1,
      "title": "Role",
      "type": "array"
    },
    "subscription": {
      "anyOf": [
        {
          "type": "string"
        },
        {
          "type": "null"
        }
      ],
      "default": null,
      "title": "Subscription"
    }
  },
  "required": [
    "name"
  ],
  "title": "UserSearchItem",
  "type": "object"
}

Fields:

email pydantic-field 

email: EmailStr | None = None

id pydantic-field 

id: UserId | None = None

last_login_time pydantic-field 

last_login_time: datetime | None = None

location pydantic-field 

location: str | None = None

location_id pydantic-field 

location_id: str | None = None

name pydantic-field 

name: str

roles pydantic-field 

roles: list[UserSearchRoleItem]

subscription pydantic-field 

subscription: str | None = None

user_class pydantic-field 

user_class: UserClass = STANDARD