User

UserDAO

Class for accessing user table.

Source code in hestia/db/dao/user.py

class UserDAO:
    """Class for accessing user table."""

    def __init__(self, session: AsyncSession = Depends(get_db_session)):
        self.session = session

    async def create_user_model(self, user_dto: UserModelDTO) -> None:
        """
        Add single user to session.

        :param user_dto: user data.
        """
        self.session.add(User(**user_dto.dict()))

    async def get_all_users(self, limit: int, offset: int) -> List[User]:
        """
        Get all User models with limit/offset pagination.

        :param limit: limit of dummies.
        :param offset: offset of dummies.
        :return: stream of dummies.
        """
        raw_users = await self.session.execute(
            select(User).limit(limit).offset(offset),
        )
        return list(raw_users.scalars().fetchall())

    async def filter(
        self,
        username: Optional[str] = None,
    ) -> List[User]:
        """
        Get specific user model.

        :param name: name of username instance.
        :return: user models.
        """
        query = select(User)
        if username:
            query = query.where(User.username == username)
        rows = await self.session.execute(query)
        return list(rows.scalars().fetchall())

create_user_model(user_dto) async

Add single user to session.

Parameters:

Name	Type	Description	Default
user_dto	UserModelDTO	user data.	required

Source code in hestia/db/dao/user.py

async def create_user_model(self, user_dto: UserModelDTO) -> None:
    """
    Add single user to session.

    :param user_dto: user data.
    """
    self.session.add(User(**user_dto.dict()))

filter(username=None) async

Get specific user model.

Parameters:

Name	Type	Description	Default
name		name of username instance.	required

Returns:

Type	Description
List[User]	user models.

Source code in hestia/db/dao/user.py

async def filter(
    self,
    username: Optional[str] = None,
) -> List[User]:
    """
    Get specific user model.

    :param name: name of username instance.
    :return: user models.
    """
    query = select(User)
    if username:
        query = query.where(User.username == username)
    rows = await self.session.execute(query)
    return list(rows.scalars().fetchall())

get_all_users(limit, offset) async

Get all User models with limit/offset pagination.

Parameters:

Name	Type	Description	Default
limit	int	limit of dummies.	required
offset	int	offset of dummies.	required

Returns:

Type	Description
List[User]	stream of dummies.

Source code in hestia/db/dao/user.py

async def get_all_users(self, limit: int, offset: int) -> List[User]:
    """
    Get all User models with limit/offset pagination.

    :param limit: limit of dummies.
    :param offset: offset of dummies.
    :return: stream of dummies.
    """
    raw_users = await self.session.execute(
        select(User).limit(limit).offset(offset),
    )
    return list(raw_users.scalars().fetchall())