Skip to content

Entity Types (🧪Beta)

albert.collections.entity_types.EntityTypeCollection

EntityTypeCollection(*, session: AlbertSession)

Bases: BaseCollection

A collection of configurable entity types in the Albert system.

Beta Feature!

Please do not use in production or without explicit guidance from Albert. You might otherwise have a bad experience. This feature currently falls outside of the Albert support contract, but we'd love your feedback!

Methods:

Name Description
get_by_id

Get an entity type by its ID.

create

Create an entity type.

update

Update an entity type.

delete

Delete an entity type.

get_rules

Get the rules for an entity type.

set_rules

Create or update the rules for an entity type.

delete_rules

Delete the rules for an entity type.

get_all

Searches for EntityType items based on the provided parameters.

Attributes:

Name Type Description
base_path
Source code in src/albert/collections/entity_types.py
def __init__(self, *, session: AlbertSession):
    """Initialize the EntityTypeCollection with the provided session."""
    super().__init__(session=session)
    self.base_path = f"/api/{EntityTypeCollection._api_version}/entitytypes"

base_path

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

get_by_id

get_by_id(*, id: EntityTypeId) -> EntityType

Get an entity type by its ID.

Parameters:

Name Type Description Default
id EntityTypeId

The ID of the entity type to get.

required
Source code in src/albert/collections/entity_types.py
def get_by_id(self, *, id: EntityTypeId) -> EntityType:
    """Get an entity type by its ID.
    Parameters
    ----------
    id : EntityTypeId
        The ID of the entity type to get.
    """
    response = self.session.get(f"{self.base_path}/{id}")
    return EntityType(**response.json())

create

create(*, entity_type: EntityType) -> EntityType

Create an entity type.

Parameters:

Name Type Description Default
entity_type EntityType

The entity type to create.

required
Source code in src/albert/collections/entity_types.py
def create(self, *, entity_type: EntityType) -> EntityType:
    """Create an entity type.
    Parameters
    ----------
    entity_type : EntityType
        The entity type to create.
    """
    response = self.session.post(
        self.base_path, json=entity_type.model_dump(by_alias=True, exclude_none=True)
    )
    return EntityType(**response.json())

update

update(*, entity_type: EntityType) -> EntityType

Update an entity type.

Parameters:

Name Type Description Default
entity_type EntityType

The entity type to update.

required
Source code in src/albert/collections/entity_types.py
def update(self, *, entity_type: EntityType) -> EntityType:
    """Update an entity type.
    Parameters
    ----------
    entity_type : EntityType
        The entity type to update.
    """
    current_entity_type = self.get_by_id(id=entity_type.id)
    patch = self._generate_patch_payload(
        existing=current_entity_type,
        updated=entity_type,
        generate_metadata_diff=False,
        stringify_values=False,
    )

    # Add special attribute updates to the patch
    special_patches = self._generate_special_attribute_patches(
        existing=current_entity_type, updated=entity_type
    )
    patch.data.extend(special_patches)

    self.session.patch(
        f"{self.base_path}/{entity_type.id}",
        json=patch.model_dump(mode="json", by_alias=True, exclude_none=True),
    )
    return self.get_by_id(id=entity_type.id)

delete

delete(*, id: EntityTypeId) -> None

Delete an entity type.

Parameters:

Name Type Description Default
id EntityTypeId

The ID of the entity type to delete.

required
Source code in src/albert/collections/entity_types.py
def delete(self, *, id: EntityTypeId) -> None:
    """Delete an entity type.
    Parameters
    ----------
    id : EntityTypeId
        The ID of the entity type to delete.
    """
    self.session.delete(f"{self.base_path}/{id}")

get_rules

get_rules(*, id: EntityTypeId) -> list[EntityTypeRule]

Get the rules for an entity type.

Parameters:

Name Type Description Default
id EntityTypeId

The ID of the entity type to get the rules for.

required
Source code in src/albert/collections/entity_types.py
def get_rules(self, *, id: EntityTypeId) -> list[EntityTypeRule]:
    """Get the rules for an entity type.
    Parameters
    ----------
    id : EntityTypeId
        The ID of the entity type to get the rules for.
    """
    response = self.session.get(f"{self.base_path}/rules/{id}")
    return [EntityTypeRule(**rule) for rule in response.json()]

set_rules

set_rules(
    *, id: EntityTypeId, rules: list[EntityTypeRule]
) -> list[EntityTypeRule]

Create or update the rules for an entity type.

Parameters:

Name Type Description Default
id EntityTypeId

The ID of the entity type to update the rules for.

required
rules list[EntityTypeRule]

The rules to update.

required

Returns:

Type Description
list[EntityTypeRule]

The updated rules.

Source code in src/albert/collections/entity_types.py
def set_rules(self, *, id: EntityTypeId, rules: list[EntityTypeRule]) -> list[EntityTypeRule]:
    """Create or update the rules for an entity type.
    Parameters
    ----------
    id : EntityTypeId
        The ID of the entity type to update the rules for.
    rules : list[EntityTypeRule]
        The rules to update.
    Returns
    -------
    list[EntityTypeRule]
        The updated rules.
    """
    response = self.session.put(
        f"{self.base_path}/rules/{id}",
        json=[rule.model_dump(exclude_none=True, by_alias=True) for rule in rules],
    )
    return [EntityTypeRule(**rule) for rule in response.json()]

delete_rules

delete_rules(*, id: EntityTypeId) -> None

Delete the rules for an entity type.

Parameters:

Name Type Description Default
id EntityTypeId

The ID of the entity type to delete the rules for.

required
Source code in src/albert/collections/entity_types.py
def delete_rules(self, *, id: EntityTypeId) -> None:
    """Delete the rules for an entity type.
    Parameters
    ----------
    id : EntityTypeId
        The ID of the entity type to delete the rules for.
    """
    self.session.delete(f"{self.base_path}/rules/{id}")

get_all

get_all(
    *,
    service: EntityServiceType | None = None,
    start_key: str | None = None,
    order: OrderBy | None = None,
    max_items: int | None = None,
) -> Iterator[EntityType]

Searches for EntityType items based on the provided parameters.

Parameters:

Name Type Description Default
service EntityServiceType | None

The service type the entity type is associated with, by default None

None
limit int

Maximum number of results to return, by default 100

required
start_key str | None

Key to start pagination from, by default None

None
order OrderBy | None

Sort order (ascending/descending), by default None

None

Yields:

Type Description
Iterator[EntityType]

Returns an iterator of EntityType items matching the search criteria.

Source code in src/albert/collections/entity_types.py
def get_all(
    self,
    *,
    service: EntityServiceType | None = None,
    start_key: str | None = None,
    order: OrderBy | None = None,
    max_items: int | None = None,
) -> Iterator[EntityType]:
    """Searches for EntityType items based on the provided parameters.
    Parameters
    ----------
    service : EntityServiceType | None, optional
        The service type the entity type is associated with, by default None
    limit : int, optional
        Maximum number of results to return, by default 100
    start_key : str | None, optional
        Key to start pagination from, by default None
    order : OrderBy | None, optional
        Sort order (ascending/descending), by default None
    Yields
    ------
    Iterator[EntityType]
        Returns an iterator of EntityType items matching the search criteria.
    """
    params = {
        "service": service.value if service else None,
        "limit": max_items,
        "startKey": start_key,
        "orderBy": order.value if order else None,
    }
    return AlbertPaginator(
        mode=PaginationMode.KEY,
        path=self.base_path,
        params=params,
        session=self.session,
        deserialize=lambda items: [EntityType(**item) for item in items],
        max_items=max_items,
    )