DB

eunomia.db.crud

create_entity(entity, db)

Create a new entity in the database.

This function creates a new entity record and its associated attributes in the database.

Parameters:

Name	Type	Description	Default
entity	EntityCreate	Pydantic model containing the entity data to be created.	required
db	Session	SQLAlchemy database session.	required

Returns:

Type	Description
Entity	The created entity as a SQLAlchemy model.

Source code in src/eunomia/db/crud.py

def create_entity(entity: schemas.EntityCreate, db: Session) -> models.Entity:
    """
    Create a new entity in the database.

    This function creates a new entity record and its associated attributes
    in the database.

    Parameters
    ----------
    entity : schemas.EntityCreate
        Pydantic model containing the entity data to be created.
    db : Session
        SQLAlchemy database session.

    Returns
    -------
    models.Entity
        The created entity as a SQLAlchemy model.
    """
    db_entity = models.Entity(
        uri=entity.uri,
        type=entity.type,
    )
    for attribute in entity.attributes:
        db_attribute = models.Attribute(key=attribute.key, value=attribute.value)
        db_entity.attributes.append(db_attribute)

    db.add(db_entity)
    db.commit()
    db.refresh(db_entity)
    return db_entity

delete_entity(db_entity, db)

Delete an entity from the database.

This function deletes an entity record and its associated attributes from the database.

Parameters:

Name	Type	Description	Default
db_entity	Entity	The entity to delete.	required
db	Session	SQLAlchemy database session.	required

Source code in src/eunomia/db/crud.py

def delete_entity(db_entity: models.Entity, db: Session) -> None:
    """
    Delete an entity from the database.

    This function deletes an entity record and its associated attributes
    from the database.

    Parameters
    ----------
    db_entity : models.Entity
        The entity to delete.
    db : Session
        SQLAlchemy database session.
    """
    delete_entity_attributes(db_entity, db)
    db.delete(db_entity)
    db.commit()
    return

delete_entity_attributes(db_entity, db)

Delete all attributes of an entity.

Parameters:

Name	Type	Description	Default
db_entity	Entity	The entity to delete attributes from.	required
db	Session	SQLAlchemy database session.	required

Source code in src/eunomia/db/crud.py

def delete_entity_attributes(db_entity: models.Entity, db: Session) -> None:
    """
    Delete all attributes of an entity.

    Parameters
    ----------
    db_entity : models.Entity
        The entity to delete attributes from.
    db : Session
        SQLAlchemy database session.
    """
    db.query(models.Attribute).filter(
        models.Attribute.entity_uri == db_entity.uri
    ).delete()
    db.commit()

get_attribute(uri, key, db)

Retrieve an attribute from the database by its key and entity uri.

Parameters:

Name	Type	Description	Default
uri	str	The uri of the entity.	required
key	str	The key of the attribute.	required
db	Session	SQLAlchemy database session.	required

Returns:

Type	Description
Attribute | None	The attribute as a SQLAlchemy model or None if it does not exist.

Source code in src/eunomia/db/crud.py

def get_attribute(uri: str, key: str, db: Session) -> models.Attribute | None:
    """
    Retrieve an attribute from the database by its key and entity uri.

    Parameters
    ----------
    uri : str
        The uri of the entity.
    key : str
        The key of the attribute.
    db : Session
        SQLAlchemy database session.

    Returns
    -------
    models.Attribute | None
        The attribute as a SQLAlchemy model or None if it does not exist.
    """
    return (
        db.query(models.Attribute)
        .filter(models.Attribute.entity_uri == uri, models.Attribute.key == key)
        .first()
    )

get_entity(uri, db)

Retrieve an entity from the database by its unique identifier.

Parameters:

Name	Type	Description	Default
uri	str	Unique identifier of the entity.	required
db	Session	SQLAlchemy database session.	required

Returns:

Type	Description
Entity | None	The entity as a SQLAlchemy model or None if it does not exist.

Source code in src/eunomia/db/crud.py

def get_entity(uri: str, db: Session) -> models.Entity | None:
    """
    Retrieve an entity from the database by its unique identifier.

    Parameters
    ----------
    uri : str
        Unique identifier of the entity.
    db : Session
        SQLAlchemy database session.

    Returns
    -------
    models.Entity | None
        The entity as a SQLAlchemy model or None if it does not exist.
    """
    return db.query(models.Entity).filter(models.Entity.uri == uri).first()

get_entity_attributes(uri, db)

Retrieve attributes for a resource by its unique identifier.

This function retrieves all attributes associated with a specific resource and returns it as a dictionary. If no entity with the specified uri is found, an empty dictionary is returned.

Parameters:

Name	Type	Description	Default
uri	str	Unique identifier of the entity.	required
db	Session	SQLAlchemy database session.	required

Returns:

Type	Description
dict	Dictionary containing all attributes key-value pairs for the entity.

Source code in src/eunomia/db/crud.py

def get_entity_attributes(uri: str, db: Session) -> dict:
    """
    Retrieve attributes for a resource by its unique identifier.

    This function retrieves all attributes associated with a specific resource
    and returns it as a dictionary. If no entity with the specified uri is found,
    an empty dictionary is returned.

    Parameters
    ----------
    uri : str
        Unique identifier of the entity.
    db : Session
        SQLAlchemy database session.

    Returns
    -------
    dict
        Dictionary containing all attributes key-value pairs for the entity.
    """
    db_entity = get_entity(uri, db)
    if db_entity is None:
        return {}
    return {attribute.key: attribute.value for attribute in db_entity.attributes}

update_entity_attributes(db_entity, attributes, db)

Update the attributes of an existing entity.

This function updates the attributes of an existing entity. If an attribute does not exist, it is created. If an attribute exists, it is updated.

Parameters:

Name	Type	Description	Default
db_entity	Entity	The entity to update.	required
attributes	list[Attribute]	The attributes to update.	required
db	Session	SQLAlchemy database session.	required

Returns:

Type	Description
Entity	The updated entity as a SQLAlchemy model.

Source code in src/eunomia/db/crud.py

def update_entity_attributes(
    db_entity: models.Entity, attributes: list[schemas.Attribute], db: Session
) -> models.Entity:
    """
    Update the attributes of an existing entity.

    This function updates the attributes of an existing entity.
    If an attribute does not exist, it is created.
    If an attribute exists, it is updated.

    Parameters
    ----------
    db_entity : models.Entity
        The entity to update.
    attributes : list[schemas.Attribute]
        The attributes to update.
    db : Session
        SQLAlchemy database session.

    Returns
    -------
    models.Entity
        The updated entity as a SQLAlchemy model.
    """
    for attribute in attributes:
        db_attribute = get_attribute(db_entity.uri, attribute.key, db)
        if db_attribute is not None:
            db_attribute.value = attribute.value
        else:
            db_attribute = models.Attribute(key=attribute.key, value=attribute.value)
            db_entity.attributes.append(db_attribute)

    db.commit()
    db.refresh(db_entity)
    return db_entity