Users

albert.collections.users

AlbertHTTPError

AlbertHTTPError(response: Response)

Bases: AlbertException

Base class for all erors due to HTTP responses.

Source code in src/albert/exceptions.py

def __init__(self, response: requests.Response):
    message = self._format_message(response)
    super().__init__(message)
    self.response = response

response instance-attribute

response = response

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

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

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_by_id	Retrieves a User by its ID.
get_current_user	Retrieves the current authenticated user.
list	Lists Users based on criteria
update	Update a User entity.

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

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

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_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

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

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"])

list

list(
    *,
    limit: int = 50,
    offset: int | None = None,
    text: str | None = None,
    status: Status | None = None,
    search_fields: str | None = None,
) -> Iterator[User]

Lists Users based on criteria

Parameters:

Name	Type	Description	Default
text	Optional[str]	text to search against, by default None	None

Returns:

Type	Description
Generator	Generator of matching Users or None

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

def list(
    self,
    *,
    limit: int = 50,
    offset: int | None = None,
    text: str | None = None,
    status: Status | None = None,
    search_fields: str | None = None,
) -> Iterator[User]:
    """Lists Users based on criteria

    Parameters
    ----------
    text : Optional[str], optional
        text to search against, by default None

    Returns
    -------
    Generator
        Generator of matching Users or None
    """

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

    params = {
        "limit": limit,
        "offset": offset,
        "status": status,
        "text": text,
        "searchFields": search_fields,
    }

    return AlbertPaginator(
        mode=PaginationMode.OFFSET,
        path=f"{self.base_path}/search",
        session=self.session,
        params=params,
        deserialize=deserialize,
    )

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

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