models

Module with database tables definitions.

Helper classes and utility functions for data management are defined here.

Modules:

Name	Description
auth	Module with classes and functions used for authentication and password handling.

Classes:

Name	Description
Credentials	Table with users credentials, used only if basic authentication is active.
Encrypted	Allows storing and retrieving password hashes using PasswordHash.
Flags	Table with global flags used by Data-Lunch.
Menu	Table with menu items.
Orders	Table with items that belongs to an order.
Password	Allows storing and retrieving password hashes using PasswordHash.
PrivilegedUsers	Table with user that have privileges (normal users and admin).
Stats	Table with number of users that ate a lunch, grouped by guest type.
Users	Table with users that placed an order.

Functions:

Name	Description
create_database	Function to create the database through SQLAlchemy models.
create_engine	Factory function for SQLAlchemy engine.
create_exclusive_session	Factory function for database exclusive session.
create_session	Factory function for database session.
get_db_dialect	Return database type (postgresql, sqlite, etc.) based on the database object passed as input.
get_flag	Get the value of a flag.
session_add_with_upsert	Use an upsert statement for postgresql to add a new record to a table,
set_flag	Set a key,value pair inside flag table.
set_sqlite_pragma	Force foreign key constraints for sqlite connections.

Attributes:

Name	Type	Description
Data	DeclarativeMeta	SQLAlchemy declarative base.
SCHEMA	str	Schema name from environment (may be overridden by configuration files).
log	Logger	Module logger.
metadata_obj	MetaData	Database metadata (SQLAlchemy).

Data module-attribute

Data: DeclarativeMeta = declarative_base(
    metadata=metadata_obj
)

SQLAlchemy declarative base.

SCHEMA module-attribute

SCHEMA: str = get('DATA_LUNCH_DB_SCHEMA', None)

Schema name from environment (may be overridden by configuration files).

log module-attribute

log: Logger = getLogger(__name__)

Module logger.

metadata_obj module-attribute

metadata_obj: MetaData = MetaData(schema=SCHEMA)

Database metadata (SQLAlchemy).

Credentials

Bases: Data

Table with users credentials, used only if basic authentication is active.

Methods:

Name	Description
__repr__	Simple object representation.
clear	Clear table and return deleted rows.
read_as_df	Read table as pandas DataFrame.

Attributes:

Name	Type	Description
__tablename__	str	Name of the table.
password_encrypted	Column	Encryped password.
password_hash	Column	Hashed password.
user	Column	Username.

Source code in dlunch/models.py

class Credentials(Data):
    """Table with users credentials, used only if basic authentication is active."""

    __tablename__: str = "credentials"
    """Name of the table."""
    user: Column = Column(
        String(100),
        primary_key=True,
        sqlite_on_conflict_primary_key="REPLACE",
    )
    """Username."""
    password_hash: Column = Column(Password(150), unique=False, nullable=False)
    """Hashed password."""
    password_encrypted: Column = Column(
        Encrypted(150),
        unique=False,
        nullable=True,
        default=None,
        server_default=None,
    )
    """Encryped password.

    Used only if basic authentication and guest users are both active."""

    @classmethod
    def clear(self, config: DictConfig) -> int:
        """Clear table and return deleted rows.

        Args:
            config (DictConfig): Hydra configuration dictionary.

        Returns:
            int: deleted rows.
        """

        session = create_session(config)
        with session:
            # Clean menu
            num_rows_deleted = session.execute(delete(self))
            session.commit()
            log.info(
                f"deleted {num_rows_deleted.rowcount} rows from table '{self.__tablename__}'"
            )

        return num_rows_deleted.rowcount

    @classmethod
    def read_as_df(self, config: DictConfig, **kwargs) -> pd.DataFrame:
        """Read table as pandas DataFrame.

        Args:
            config (DictConfig): Hydra configuration dictionary.

        Returns:
            pd.DataFrame: dataframe with table content.
        """
        df = pd.read_sql_table(
            self.__tablename__,
            create_engine(config=config),
            schema=config.db.get("schema", SCHEMA),
            **kwargs,
        )

        return df

    def __repr__(self) -> str:
        """Simple object representation.

        Returns:
            str: string representation.
        """
        return f"<CREDENTIAL:{self.user}>"

    @validates("password_hash")
    def _validate_password(
        self, key: str, password: str
    ) -> auth.PasswordHash | None:
        """Function that validate password input.

        It converts string to auth.PasswordHash if necessary.

        Args:
            key (str): validated attribute name.
            password (str): hashed password to be validated.

        Returns:
            auth.PasswordHash | None: validated hashed password.
        """
        return getattr(type(self), key).type.validator(password)

    @validates("password_encrypted")
    def _validate_encrypted(
        self, key: str, password: str
    ) -> auth.PasswordEncrypt | None:
        """Function that validate encrypted input.

        It converts string to auth.PasswordEncrypt if necessary.

        Args:
            key (str): validated attribute name.
            password (str): encrypted password to be validated.

        Returns:
            auth.PasswordEncrypt | None: validated encrypted password.
        """
        return getattr(type(self), key).type.validator(password)

__tablename__ class-attribute instance-attribute

__tablename__: str = 'credentials'

Name of the table.

password_encrypted class-attribute instance-attribute

password_encrypted: Column = Column(
    Encrypted(150),
    unique=False,
    nullable=True,
    default=None,
    server_default=None,
)

Encryped password.

Used only if basic authentication and guest users are both active.

password_hash class-attribute instance-attribute

password_hash: Column = Column(
    Password(150), unique=False, nullable=False
)

Hashed password.

user class-attribute instance-attribute

user: Column = Column(
    String(100),
    primary_key=True,
    sqlite_on_conflict_primary_key="REPLACE",
)

Username.

__repr__

__repr__() -> str

Simple object representation.

Returns:

Type	Description
str	string representation.

Source code in dlunch/models.py

def __repr__(self) -> str:
    """Simple object representation.

    Returns:
        str: string representation.
    """
    return f"<CREDENTIAL:{self.user}>"

clear classmethod

clear(config: DictConfig) -> int

Clear table and return deleted rows.

Parameters:

Name	Type	Description	Default
config	DictConfig	Hydra configuration dictionary.	required

Returns:

Type	Description
int	deleted rows.

Source code in dlunch/models.py

@classmethod
def clear(self, config: DictConfig) -> int:
    """Clear table and return deleted rows.

    Args:
        config (DictConfig): Hydra configuration dictionary.

    Returns:
        int: deleted rows.
    """

    session = create_session(config)
    with session:
        # Clean menu
        num_rows_deleted = session.execute(delete(self))
        session.commit()
        log.info(
            f"deleted {num_rows_deleted.rowcount} rows from table '{self.__tablename__}'"
        )

    return num_rows_deleted.rowcount

read_as_df classmethod

read_as_df(config: DictConfig, **kwargs) -> DataFrame

Read table as pandas DataFrame.

Parameters:

Name	Type	Description	Default
config	DictConfig	Hydra configuration dictionary.	required

Returns:

Type	Description
DataFrame	dataframe with table content.

Source code in dlunch/models.py

@classmethod
def read_as_df(self, config: DictConfig, **kwargs) -> pd.DataFrame:
    """Read table as pandas DataFrame.

    Args:
        config (DictConfig): Hydra configuration dictionary.

    Returns:
        pd.DataFrame: dataframe with table content.
    """
    df = pd.read_sql_table(
        self.__tablename__,
        create_engine(config=config),
        schema=config.db.get("schema", SCHEMA),
        **kwargs,
    )

    return df

Encrypted

Bases: TypeDecorator

Allows storing and retrieving password hashes using PasswordHash.

Methods:

Name	Description
process_bind_param	Ensure the value is a PasswordEncrypt and then return the encrypted password.
process_result_value	Convert the hash to a PasswordEncrypt, if it's non-NULL.
validator	Provides a validator/converter used by @validates.

Attributes:

Name	Type	Description
impl	String	Base column implementation.

Source code in dlunch/models.py

class Encrypted(TypeDecorator):
    """Allows storing and retrieving password hashes using PasswordHash."""

    impl: String = String
    """Base column implementation."""

    def process_bind_param(
        self, value: auth.PasswordEncrypt | str | None, dialect
    ) -> str | None:
        """Ensure the value is a PasswordEncrypt and then return the encrypted password.

        Args:
            value (auth.PasswordEncrypt | str | None): input value (plain password or encrypted or `None` if empty)
            dialect (Any): dialect (not used).

        Returns:
            str | None: encrypted password or `None` if empty.
        """
        converted_value = self._convert(value)
        if converted_value:
            return converted_value.encrypted_password
        else:
            return None

    def process_result_value(
        self, value: str | None, dialect
    ) -> auth.PasswordEncrypt | None:
        """Convert the hash to a PasswordEncrypt, if it's non-NULL.

        Args:
            value (str | None): input value (plain password or encrypted or `None` if empty)
            dialect (Any): dialect (not used).

        Returns:
            auth.PasswordEncrypt | None: encrypted password as object or `None` (if nothing is passed as value).
        """
        if value is not None:
            return auth.PasswordEncrypt(value)

    def validator(
        self, password: auth.PasswordEncrypt | str | None
    ) -> auth.PasswordEncrypt | None:
        """Provides a validator/converter used by @validates.

        Args:
            password (auth.PasswordEncrypt | str | None): input value (plain password or encrypted or `None` if empty)

        Returns:
            auth.PasswordEncrypt | None: encrypted password as object or `None` (if nothing is passed as value).
        """
        return self._convert(password)

    def _convert(
        self, value: auth.PasswordEncrypt | str | None
    ) -> auth.PasswordEncrypt | None:
        """Returns a PasswordEncrypt from the given string.

        PasswordEncrypt instances or None values will return unchanged.
        Strings will be encrypted and the resulting PasswordEncrypt returned.
        Any other input will result in a TypeError.

        Args:
            value (auth.PasswordEncrypt | str | None): input value (plain password or encrypted or `None` if empty)

        Raises:
            TypeError: unknown type.

        Returns:
            auth.PasswordEncrypt | None: encrypted password as object or `None` (if nothing is passed as value).
        """
        if isinstance(value, auth.PasswordEncrypt):
            return value
        elif isinstance(value, str):
            return auth.PasswordEncrypt.from_str(value)
        elif value is not None:
            raise TypeError(
                f"Cannot initialize PasswordEncrypt from type '{type(value)}'"
            )

        # Reached only if value is None
        return None

impl class-attribute instance-attribute

impl: String = String

Base column implementation.

process_bind_param

process_bind_param(
    value: PasswordEncrypt | str | None, dialect
) -> str | None

Ensure the value is a PasswordEncrypt and then return the encrypted password.

Parameters:

Name	Type	Description	Default
value	PasswordEncrypt | str | None	input value (plain password or encrypted or None if empty)	required
dialect	Any	dialect (not used).	required

Returns:

Type	Description
str | None	encrypted password or None if empty.

Source code in dlunch/models.py

def process_bind_param(
    self, value: auth.PasswordEncrypt | str | None, dialect
) -> str | None:
    """Ensure the value is a PasswordEncrypt and then return the encrypted password.

    Args:
        value (auth.PasswordEncrypt | str | None): input value (plain password or encrypted or `None` if empty)
        dialect (Any): dialect (not used).

    Returns:
        str | None: encrypted password or `None` if empty.
    """
    converted_value = self._convert(value)
    if converted_value:
        return converted_value.encrypted_password
    else:
        return None

process_result_value

process_result_value(
    value: str | None, dialect
) -> PasswordEncrypt | None

Convert the hash to a PasswordEncrypt, if it's non-NULL.

Parameters:

Name	Type	Description	Default
value	str | None	input value (plain password or encrypted or None if empty)	required
dialect	Any	dialect (not used).	required

Returns:

Type	Description
PasswordEncrypt | None	encrypted password as object or None (if nothing is passed as value).

Source code in dlunch/models.py

def process_result_value(
    self, value: str | None, dialect
) -> auth.PasswordEncrypt | None:
    """Convert the hash to a PasswordEncrypt, if it's non-NULL.

    Args:
        value (str | None): input value (plain password or encrypted or `None` if empty)
        dialect (Any): dialect (not used).

    Returns:
        auth.PasswordEncrypt | None: encrypted password as object or `None` (if nothing is passed as value).
    """
    if value is not None:
        return auth.PasswordEncrypt(value)

validator

validator(
    password: PasswordEncrypt | str | None,
) -> PasswordEncrypt | None

Provides a validator/converter used by @validates.

Parameters:

Name	Type	Description	Default
password	PasswordEncrypt | str | None	input value (plain password or encrypted or None if empty)	required

Returns:

Type	Description
PasswordEncrypt | None	encrypted password as object or None (if nothing is passed as value).

Source code in dlunch/models.py

def validator(
    self, password: auth.PasswordEncrypt | str | None
) -> auth.PasswordEncrypt | None:
    """Provides a validator/converter used by @validates.

    Args:
        password (auth.PasswordEncrypt | str | None): input value (plain password or encrypted or `None` if empty)

    Returns:
        auth.PasswordEncrypt | None: encrypted password as object or `None` (if nothing is passed as value).
    """
    return self._convert(password)

Flags

Bases: Data

Table with global flags used by Data-Lunch.

'No more orders' flag and guest override flags are stored here.

Methods:

Name	Description
__repr__	Simple object representation.
clear	Clear table and return deleted rows.
clear_guest_override	Clear 'guest_override' flags and return deleted rows
read_as_df	Read table as pandas DataFrame.

Attributes:

Name	Type	Description
__tablename__	str	Name of the table.
id	Column	Flag ID (name).
value	Column	Flag value.

Source code in dlunch/models.py

class Flags(Data):
    """Table with global flags used by Data-Lunch.

    'No more orders' flag and guest override flags are stored here.
    """

    __tablename__: str = "flags"
    """Name of the table."""
    id: Column = Column(
        String(50),
        primary_key=True,
        nullable=False,
        sqlite_on_conflict_primary_key="REPLACE",
    )
    """Flag ID (name)."""
    value: Column = Column(Boolean, nullable=False)
    """Flag value."""

    @classmethod
    def clear(self, config: DictConfig) -> int:
        """Clear table and return deleted rows.

        Args:
            config (DictConfig): Hydra configuration dictionary.

        Returns:
            int: deleted rows.
        """

        session = create_session(config)
        with session:
            # Clean menu
            num_rows_deleted = session.execute(delete(self))
            session.commit()
            log.info(
                f"deleted {num_rows_deleted.rowcount} rows from table '{self.__tablename__}'"
            )

        return num_rows_deleted.rowcount

    @classmethod
    def clear_guest_override(self, config: DictConfig) -> int:
        """Clear 'guest_override' flags and return deleted rows

        Args:
            config (DictConfig): Hydra configuration dictionary.

        Returns:
            int: deleted rows.
        """

        session = create_session(config)
        with session:
            # Clean menu
            num_rows_deleted = session.execute(
                delete(self).where(self.id.like("%_guest_override"))
            )
            session.commit()
            log.info(
                f"deleted {num_rows_deleted.rowcount} rows (guest override) from table '{self.__tablename__}'"
            )

        return num_rows_deleted.rowcount

    @classmethod
    def read_as_df(self, config: DictConfig, **kwargs) -> pd.DataFrame:
        """Read table as pandas DataFrame.

        Args:
            config (DictConfig): Hydra configuration dictionary.

        Returns:
            pd.DataFrame: dataframe with table content.
        """
        df = pd.read_sql_table(
            self.__tablename__,
            create_engine(config=config),
            schema=config.db.get("schema", SCHEMA),
            **kwargs,
        )

        return df

    def __repr__(self) -> str:
        """Simple object representation.

        Returns:
            str: string representation.
        """
        return f"<FLAG:{self.id} - value:{self.value}>"

__tablename__ class-attribute instance-attribute

__tablename__: str = 'flags'

Name of the table.

id class-attribute instance-attribute

id: Column = Column(
    String(50),
    primary_key=True,
    nullable=False,
    sqlite_on_conflict_primary_key="REPLACE",
)

Flag ID (name).

value class-attribute instance-attribute

value: Column = Column(Boolean, nullable=False)

Flag value.

__repr__

__repr__() -> str

Simple object representation.

Returns:

Type	Description
str	string representation.

Source code in dlunch/models.py

def __repr__(self) -> str:
    """Simple object representation.

    Returns:
        str: string representation.
    """
    return f"<FLAG:{self.id} - value:{self.value}>"

clear classmethod

clear(config: DictConfig) -> int

Clear table and return deleted rows.

Parameters:

Name	Type	Description	Default
config	DictConfig	Hydra configuration dictionary.	required

Returns:

Type	Description
int	deleted rows.

Source code in dlunch/models.py

@classmethod
def clear(self, config: DictConfig) -> int:
    """Clear table and return deleted rows.

    Args:
        config (DictConfig): Hydra configuration dictionary.

    Returns:
        int: deleted rows.
    """

    session = create_session(config)
    with session:
        # Clean menu
        num_rows_deleted = session.execute(delete(self))
        session.commit()
        log.info(
            f"deleted {num_rows_deleted.rowcount} rows from table '{self.__tablename__}'"
        )

    return num_rows_deleted.rowcount

clear_guest_override classmethod

clear_guest_override(config: DictConfig) -> int

Clear 'guest_override' flags and return deleted rows

Parameters:

Name	Type	Description	Default
config	DictConfig	Hydra configuration dictionary.	required

Returns:

Type	Description
int	deleted rows.

Source code in dlunch/models.py

@classmethod
def clear_guest_override(self, config: DictConfig) -> int:
    """Clear 'guest_override' flags and return deleted rows

    Args:
        config (DictConfig): Hydra configuration dictionary.

    Returns:
        int: deleted rows.
    """

    session = create_session(config)
    with session:
        # Clean menu
        num_rows_deleted = session.execute(
            delete(self).where(self.id.like("%_guest_override"))
        )
        session.commit()
        log.info(
            f"deleted {num_rows_deleted.rowcount} rows (guest override) from table '{self.__tablename__}'"
        )

    return num_rows_deleted.rowcount

read_as_df classmethod

read_as_df(config: DictConfig, **kwargs) -> DataFrame

Read table as pandas DataFrame.

Parameters:

Name	Type	Description	Default
config	DictConfig	Hydra configuration dictionary.	required

Returns:

Type	Description
DataFrame	dataframe with table content.

Source code in dlunch/models.py

@classmethod
def read_as_df(self, config: DictConfig, **kwargs) -> pd.DataFrame:
    """Read table as pandas DataFrame.

    Args:
        config (DictConfig): Hydra configuration dictionary.

    Returns:
        pd.DataFrame: dataframe with table content.
    """
    df = pd.read_sql_table(
        self.__tablename__,
        create_engine(config=config),
        schema=config.db.get("schema", SCHEMA),
        **kwargs,
    )

    return df

Menu

Bases: Data

Table with menu items.

Methods:

Name	Description
__repr__	Simple object representation.
clear	Clear table and return deleted rows.
read_as_df	Read table as pandas DataFrame.

Attributes:

Name	Type	Description
__tablename__	str	Name of the table.
id	Column	Menu item ID.
item	Column	Item name.
orders	DeclarativeMeta	Orders connected to each menu item.

Source code in dlunch/models.py

class Menu(Data):
    """Table with menu items."""

    __tablename__: str = "menu"
    """Name of the table."""
    id: Column = Column(
        Integer, Identity(start=1, cycle=True), primary_key=True
    )
    """Menu item ID."""
    item: Column = Column(String(250), unique=False, nullable=False)
    """Item name."""
    orders: DeclarativeMeta = relationship(
        "Orders",
        back_populates="menu_item",
        cascade="all, delete-orphan",
        passive_deletes=True,
    )
    """Orders connected to each menu item."""

    @classmethod
    def clear(self, config: DictConfig) -> int:
        """Clear table and return deleted rows.

        Args:
            config (DictConfig): Hydra configuration dictionary.

        Returns:
            int: deleted rows.
        """

        session = create_session(config)
        with session:
            # Clean menu
            num_rows_deleted = session.execute(delete(self))
            session.commit()
            log.info(
                f"deleted {num_rows_deleted.rowcount} rows from table '{self.__tablename__}'"
            )

        return num_rows_deleted.rowcount

    @classmethod
    def read_as_df(self, config: DictConfig, **kwargs) -> pd.DataFrame:
        """Read table as pandas DataFrame.

        Args:
            config (DictConfig): Hydra configuration dictionary.

        Returns:
            pd.DataFrame: dataframe with table content.
        """
        df = pd.read_sql_table(
            self.__tablename__,
            create_engine(config=config),
            schema=config.db.get("schema", SCHEMA),
            **kwargs,
        )

        return df

    def __repr__(self) -> str:
        """Simple object representation.

        Returns:
            str: string representation.
        """
        return f"<MENU_ITEM:{self.id} - {self.item}>"

__tablename__ class-attribute instance-attribute

__tablename__: str = 'menu'

Name of the table.

id class-attribute instance-attribute

id: Column = Column(
    Integer, Identity(start=1, cycle=True), primary_key=True
)

Menu item ID.

item class-attribute instance-attribute

item: Column = Column(
    String(250), unique=False, nullable=False
)

Item name.

orders class-attribute instance-attribute

orders: DeclarativeMeta = relationship(
    "Orders",
    back_populates="menu_item",
    cascade="all, delete-orphan",
    passive_deletes=True,
)

Orders connected to each menu item.

__repr__

__repr__() -> str

Simple object representation.

Returns:

Type	Description
str	string representation.

Source code in dlunch/models.py

def __repr__(self) -> str:
    """Simple object representation.

    Returns:
        str: string representation.
    """
    return f"<MENU_ITEM:{self.id} - {self.item}>"

clear classmethod

clear(config: DictConfig) -> int

Clear table and return deleted rows.

Parameters:

Name	Type	Description	Default
config	DictConfig	Hydra configuration dictionary.	required

Returns:

Type	Description
int	deleted rows.

Source code in dlunch/models.py

@classmethod
def clear(self, config: DictConfig) -> int:
    """Clear table and return deleted rows.

    Args:
        config (DictConfig): Hydra configuration dictionary.

    Returns:
        int: deleted rows.
    """

    session = create_session(config)
    with session:
        # Clean menu
        num_rows_deleted = session.execute(delete(self))
        session.commit()
        log.info(
            f"deleted {num_rows_deleted.rowcount} rows from table '{self.__tablename__}'"
        )

    return num_rows_deleted.rowcount

read_as_df classmethod

read_as_df(config: DictConfig, **kwargs) -> DataFrame

Read table as pandas DataFrame.

Parameters:

Name	Type	Description	Default
config	DictConfig	Hydra configuration dictionary.	required

Returns:

Type	Description
DataFrame	dataframe with table content.

Source code in dlunch/models.py

@classmethod
def read_as_df(self, config: DictConfig, **kwargs) -> pd.DataFrame:
    """Read table as pandas DataFrame.

    Args:
        config (DictConfig): Hydra configuration dictionary.

    Returns:
        pd.DataFrame: dataframe with table content.
    """
    df = pd.read_sql_table(
        self.__tablename__,
        create_engine(config=config),
        schema=config.db.get("schema", SCHEMA),
        **kwargs,
    )

    return df

Orders

Bases: Data

Table with items that belongs to an order.

Methods:

Name	Description
__repr__	Simple object representation.
clear	Clear table and return deleted rows.
read_as_df	Read table as pandas DataFrame.

Attributes:

Name	Type	Description
__tablename__	str	Name of the table.
id	Column	Order ID.
menu_item	DeclarativeMeta	Menu items connected to each order (see menu table).
menu_item_id	Column	ID of the menu item included in the order.
note	Column	Note field attached to the order.
user	Column	User that placed the order.
user_record	DeclarativeMeta	User connected to this order.

Source code in dlunch/models.py

class Orders(Data):
    """Table with items that belongs to an order."""

    __tablename__: str = "orders"
    """Name of the table."""
    id: Column = Column(
        Integer, Identity(start=1, cycle=True), primary_key=True
    )
    """Order ID."""
    user: Column = Column(
        String(100),
        ForeignKey("users.id", ondelete="CASCADE"),
        index=True,
        nullable=False,
    )
    """User that placed the order."""
    user_record: DeclarativeMeta = relationship(
        "Users", back_populates="orders", uselist=False
    )
    """User connected to this order."""
    menu_item_id: Column = Column(
        Integer,
        ForeignKey("menu.id", ondelete="CASCADE"),
        nullable=False,
    )
    """ID of the menu item included in the order."""
    menu_item: DeclarativeMeta = relationship("Menu", back_populates="orders")
    """Menu items connected to each order (see `menu` table)."""
    note: Column = Column(String(300), unique=False, nullable=True)
    """Note field attached to the order."""

    @classmethod
    def clear(self, config: DictConfig) -> int:
        """Clear table and return deleted rows.

        Args:
            config (DictConfig): Hydra configuration dictionary.

        Returns:
            int: deleted rows.
        """

        session = create_session(config)
        with session:
            # Clean menu
            num_rows_deleted = session.execute(delete(self))
            session.commit()
            log.info(
                f"deleted {num_rows_deleted.rowcount} rows from table '{self.__tablename__}'"
            )

        return num_rows_deleted.rowcount

    @classmethod
    def read_as_df(self, config: DictConfig, **kwargs) -> pd.DataFrame:
        """Read table as pandas DataFrame.

        Args:
            config (DictConfig): Hydra configuration dictionary.

        Returns:
            pd.DataFrame: dataframe with table content.
        """
        df = pd.read_sql_table(
            self.__tablename__,
            create_engine(config=config),
            schema=config.db.get("schema", SCHEMA),
            **kwargs,
        )

        return df

    def __repr__(self) -> str:
        """Simple object representation.

        Returns:
            str: string representation.
        """
        return f"<ORDER:{self.user}, {self.menu_item.item}>"

__tablename__ class-attribute instance-attribute

__tablename__: str = 'orders'

Name of the table.

id class-attribute instance-attribute

id: Column = Column(
    Integer, Identity(start=1, cycle=True), primary_key=True
)

Order ID.

menu_item class-attribute instance-attribute

menu_item: DeclarativeMeta = relationship(
    "Menu", back_populates="orders"
)

Menu items connected to each order (see menu table).

menu_item_id class-attribute instance-attribute

menu_item_id: Column = Column(
    Integer,
    ForeignKey("menu.id", ondelete="CASCADE"),
    nullable=False,
)

ID of the menu item included in the order.

note class-attribute instance-attribute

note: Column = Column(
    String(300), unique=False, nullable=True
)

Note field attached to the order.

user class-attribute instance-attribute

user: Column = Column(
    String(100),
    ForeignKey("users.id", ondelete="CASCADE"),
    index=True,
    nullable=False,
)

User that placed the order.

user_record class-attribute instance-attribute

user_record: DeclarativeMeta = relationship(
    "Users", back_populates="orders", uselist=False
)

User connected to this order.

__repr__

__repr__() -> str

Simple object representation.

Returns:

Type	Description
str	string representation.

Source code in dlunch/models.py

def __repr__(self) -> str:
    """Simple object representation.

    Returns:
        str: string representation.
    """
    return f"<ORDER:{self.user}, {self.menu_item.item}>"

clear classmethod

clear(config: DictConfig) -> int

Clear table and return deleted rows.

Parameters:

Name	Type	Description	Default
config	DictConfig	Hydra configuration dictionary.	required

Returns:

Type	Description
int	deleted rows.

Source code in dlunch/models.py

@classmethod
def clear(self, config: DictConfig) -> int:
    """Clear table and return deleted rows.

    Args:
        config (DictConfig): Hydra configuration dictionary.

    Returns:
        int: deleted rows.
    """

    session = create_session(config)
    with session:
        # Clean menu
        num_rows_deleted = session.execute(delete(self))
        session.commit()
        log.info(
            f"deleted {num_rows_deleted.rowcount} rows from table '{self.__tablename__}'"
        )

    return num_rows_deleted.rowcount

read_as_df classmethod

read_as_df(config: DictConfig, **kwargs) -> DataFrame

Read table as pandas DataFrame.

Parameters:

Name	Type	Description	Default
config	DictConfig	Hydra configuration dictionary.	required

Returns:

Type	Description
DataFrame	dataframe with table content.

Source code in dlunch/models.py

@classmethod
def read_as_df(self, config: DictConfig, **kwargs) -> pd.DataFrame:
    """Read table as pandas DataFrame.

    Args:
        config (DictConfig): Hydra configuration dictionary.

    Returns:
        pd.DataFrame: dataframe with table content.
    """
    df = pd.read_sql_table(
        self.__tablename__,
        create_engine(config=config),
        schema=config.db.get("schema", SCHEMA),
        **kwargs,
    )

    return df

Password

Bases: TypeDecorator

Allows storing and retrieving password hashes using PasswordHash.

Methods:

Name	Description
process_bind_param	Ensure the value is a PasswordHash and then return its hash.
process_result_value	Convert the hash to a PasswordHash, if it's non-NULL.
validator	Provides a validator/converter used by @validates.

Attributes:

Name	Type	Description
impl	String	Base column implementation.

Source code in dlunch/models.py

class Password(TypeDecorator):
    """Allows storing and retrieving password hashes using PasswordHash."""

    impl: String = String
    """Base column implementation."""

    def process_bind_param(
        self, value: auth.PasswordHash | str | None, dialect
    ) -> str:
        """Ensure the value is a PasswordHash and then return its hash.

        Args:
            value (auth.PasswordHash | str): input value (plain password or hash, or `None` if empty).
            dialect (Any): dialect (not used).

        Returns:
            str: password hash.
        """
        return self._convert(value).hashed_password

    def process_result_value(
        self, value: str | None, dialect
    ) -> auth.PasswordHash | None:
        """Convert the hash to a PasswordHash, if it's non-NULL.

        Args:
            value (str | None): password hash (or `None` if empty).
            dialect (Any): dialect (not used).

        Returns:
            auth.PasswordHash | None: hashed password as object or `None` (if nothing is passed as value).
        """
        if value is not None:
            return auth.PasswordHash(value)

    def validator(
        self, password: auth.PasswordHash | str | None
    ) -> auth.PasswordHash | None:
        """Provides a validator/converter used by @validates.

        Args:
            password (auth.PasswordHash | str | None): input value (plain password or hash or `None` if empty).

        Returns:
            auth.PasswordHash | None: hashed password as object or `None` (if nothing is passed as value).
        """
        return self._convert(password)

    def _convert(
        self, value: auth.PasswordHash | str | None
    ) -> auth.PasswordHash | None:
        """Returns a PasswordHash from the given string.

        PasswordHash instances or None values will return unchanged.
        Strings will be hashed and the resulting PasswordHash returned.
        Any other input will result in a TypeError.

        Args:
            value (auth.PasswordHash | str | None): input value (plain password or hash or `None` if empty).

        Raises:
            TypeError: unknown type.

        Returns:
            auth.PasswordHash | None: hashed password as object or `None` (if nothing is passed as value).
        """
        if isinstance(value, auth.PasswordHash):
            return value
        elif isinstance(value, str):
            return auth.PasswordHash.from_str(value)
        elif value is not None:
            raise TypeError(
                f"Cannot initialize PasswordHash from type '{type(value)}'"
            )

        # Reached only if value is None
        return None

impl class-attribute instance-attribute

impl: String = String

Base column implementation.

process_bind_param

process_bind_param(
    value: PasswordHash | str | None, dialect
) -> str

Ensure the value is a PasswordHash and then return its hash.

Parameters:

Name	Type	Description	Default
value	PasswordHash | str	input value (plain password or hash, or None if empty).	required
dialect	Any	dialect (not used).	required

Returns:

Type	Description
str	password hash.

Source code in dlunch/models.py

def process_bind_param(
    self, value: auth.PasswordHash | str | None, dialect
) -> str:
    """Ensure the value is a PasswordHash and then return its hash.

    Args:
        value (auth.PasswordHash | str): input value (plain password or hash, or `None` if empty).
        dialect (Any): dialect (not used).

    Returns:
        str: password hash.
    """
    return self._convert(value).hashed_password

process_result_value

process_result_value(
    value: str | None, dialect
) -> PasswordHash | None

Convert the hash to a PasswordHash, if it's non-NULL.

Parameters:

Name	Type	Description	Default
value	str | None	password hash (or None if empty).	required
dialect	Any	dialect (not used).	required

Returns:

Type	Description
PasswordHash | None	hashed password as object or None (if nothing is passed as value).

Source code in dlunch/models.py

def process_result_value(
    self, value: str | None, dialect
) -> auth.PasswordHash | None:
    """Convert the hash to a PasswordHash, if it's non-NULL.

    Args:
        value (str | None): password hash (or `None` if empty).
        dialect (Any): dialect (not used).

    Returns:
        auth.PasswordHash | None: hashed password as object or `None` (if nothing is passed as value).
    """
    if value is not None:
        return auth.PasswordHash(value)

validator

validator(
    password: PasswordHash | str | None,
) -> PasswordHash | None

Provides a validator/converter used by @validates.

Parameters:

Name	Type	Description	Default
password	PasswordHash | str | None	input value (plain password or hash or None if empty).	required

Returns:

Type	Description
PasswordHash | None	hashed password as object or None (if nothing is passed as value).

Source code in dlunch/models.py

def validator(
    self, password: auth.PasswordHash | str | None
) -> auth.PasswordHash | None:
    """Provides a validator/converter used by @validates.

    Args:
        password (auth.PasswordHash | str | None): input value (plain password or hash or `None` if empty).

    Returns:
        auth.PasswordHash | None: hashed password as object or `None` (if nothing is passed as value).
    """
    return self._convert(password)

PrivilegedUsers

Bases: Data

Table with user that have privileges (normal users and admin).

If enabled guests are all the authenticated users that do not belong to this table (see config key auth.authorize_guest_users and basic_auth.guest_user)

Methods:

Name	Description
__repr__	Simple object representation.
clear	Clear table and return deleted rows.
read_as_df	Read table as pandas DataFrame.

Attributes:

Name	Type	Description
__tablename__	str	Name of the table.
admin	Column	Admin flag (true if admin).
user	Column	User name.

Source code in dlunch/models.py

class PrivilegedUsers(Data):
    """Table with user that have privileges (normal users and admin).

    If enabled guests are all the authenticated users that do not belong to this table
    (see config key `auth.authorize_guest_users` and `basic_auth.guest_user`)
    """

    __tablename__: str = "privileged_users"
    """Name of the table."""
    user: Column = Column(
        String(100),
        primary_key=True,
        sqlite_on_conflict_primary_key="REPLACE",
    )
    """User name."""
    admin: Column = Column(
        Boolean, nullable=False, default=False, server_default=sql_false()
    )
    """Admin flag (true if admin)."""

    @classmethod
    def clear(self, config: DictConfig) -> int:
        """Clear table and return deleted rows.

        Args:
            config (DictConfig): Hydra configuration dictionary.

        Returns:
            int: deleted rows.
        """

        session = create_session(config)
        with session:
            # Clean menu
            num_rows_deleted = session.execute(delete(self))
            session.commit()
            log.info(
                f"deleted {num_rows_deleted.rowcount} rows from table '{self.__tablename__}'"
            )

        return num_rows_deleted.rowcount

    @classmethod
    def read_as_df(self, config: DictConfig, **kwargs) -> pd.DataFrame:
        """Read table as pandas DataFrame.

        Args:
            config (DictConfig): Hydra configuration dictionary.

        Returns:
            pd.DataFrame: dataframe with table content.
        """
        df = pd.read_sql_table(
            self.__tablename__,
            create_engine(config=config),
            schema=config.db.get("schema", SCHEMA),
            **kwargs,
        )

        return df

    def __repr__(self) -> str:
        """Simple object representation.

        Returns:
            str: string representation.
        """
        return f"<PRIVILEGED_USER:{self.id}>"

__tablename__ class-attribute instance-attribute

__tablename__: str = 'privileged_users'

Name of the table.

admin class-attribute instance-attribute

admin: Column = Column(
    Boolean,
    nullable=False,
    default=False,
    server_default=false(),
)

Admin flag (true if admin).

user class-attribute instance-attribute

user: Column = Column(
    String(100),
    primary_key=True,
    sqlite_on_conflict_primary_key="REPLACE",
)

User name.

__repr__

__repr__() -> str

Simple object representation.

Returns:

Type	Description
str	string representation.

Source code in dlunch/models.py

def __repr__(self) -> str:
    """Simple object representation.

    Returns:
        str: string representation.
    """
    return f"<PRIVILEGED_USER:{self.id}>"

clear classmethod

clear(config: DictConfig) -> int

Clear table and return deleted rows.

Parameters:

Name	Type	Description	Default
config	DictConfig	Hydra configuration dictionary.	required

Returns:

Type	Description
int	deleted rows.

Source code in dlunch/models.py

@classmethod
def clear(self, config: DictConfig) -> int:
    """Clear table and return deleted rows.

    Args:
        config (DictConfig): Hydra configuration dictionary.

    Returns:
        int: deleted rows.
    """

    session = create_session(config)
    with session:
        # Clean menu
        num_rows_deleted = session.execute(delete(self))
        session.commit()
        log.info(
            f"deleted {num_rows_deleted.rowcount} rows from table '{self.__tablename__}'"
        )

    return num_rows_deleted.rowcount

read_as_df classmethod

read_as_df(config: DictConfig, **kwargs) -> DataFrame

Read table as pandas DataFrame.

Parameters:

Name	Type	Description	Default
config	DictConfig	Hydra configuration dictionary.	required

Returns:

Type	Description
DataFrame	dataframe with table content.

Source code in dlunch/models.py

@classmethod
def read_as_df(self, config: DictConfig, **kwargs) -> pd.DataFrame:
    """Read table as pandas DataFrame.

    Args:
        config (DictConfig): Hydra configuration dictionary.

    Returns:
        pd.DataFrame: dataframe with table content.
    """
    df = pd.read_sql_table(
        self.__tablename__,
        create_engine(config=config),
        schema=config.db.get("schema", SCHEMA),
        **kwargs,
    )

    return df

Stats

Bases: Data

Table with number of users that ate a lunch, grouped by guest type.

Methods:

Name	Description
__repr__	Simple object representation.
clear	Clear table and return deleted rows.
read_as_df	Read table as pandas DataFrame.

Attributes:

Name	Type	Description
__table_args__	tuple	Table arguments, used to create primary key (ON CONFLICT options for composite
__tablename__	str	Name of the table.
date	Column	Day to which the statistics refers to.
guest	Column	Different kind of guests are identified by the value defined in config files
hungry_people	Column	Number of people that ate in a certain day.

Source code in dlunch/models.py

class Stats(Data):
    """Table with number of users that ate a lunch, grouped by guest type."""

    # Primary key handled with __table_args__ because ON CONFLICT for composite
    # primary key is available only with __table_args__
    __tablename__: str = "stats"
    """Name of the table."""
    __table_args__: tuple = (
        PrimaryKeyConstraint(
            "date", "guest", name="stats_pkey", sqlite_on_conflict="REPLACE"
        ),
    )
    """Table arguments, used to create primary key (ON CONFLICT options for composite
    primary key is available only with __table_args__)."""
    date: Column = Column(
        Date,
        nullable=False,
        server_default=func.current_date(),
    )
    """Day to which the statistics refers to."""
    guest: Column = Column(
        String(20),
        nullable=True,
        default="NotAGuest",
        server_default="NotAGuest",
    )
    """Different kind of guests are identified by the value defined in config files
    (see config key `panel.guest_types`).
    'NotAGuest' is the value used for locals.
    """
    hungry_people: Column = Column(
        Integer, nullable=False, default=0, server_default="0"
    )
    """Number of people that ate in a certain day.
    different kind of guests are identified by the value in guest column.
    """

    @classmethod
    def clear(self, config: DictConfig) -> int:
        """Clear table and return deleted rows.

        Args:
            config (DictConfig): Hydra configuration dictionary.

        Returns:
            int: deleted rows.
        """

        session = create_session(config)
        with session:
            # Clean menu
            num_rows_deleted = session.execute(delete(self))
            session.commit()
            log.info(
                f"deleted {num_rows_deleted.rowcount} rows from table '{self.__tablename__}'"
            )

        return num_rows_deleted.rowcount

    @classmethod
    def read_as_df(self, config: DictConfig, **kwargs) -> pd.DataFrame:
        """Read table as pandas DataFrame.

        Args:
            config (DictConfig): Hydra configuration dictionary.

        Returns:
            pd.DataFrame: dataframe with table content.
        """
        df = pd.read_sql_table(
            self.__tablename__,
            create_engine(config=config),
            schema=config.db.get("schema", SCHEMA),
            **kwargs,
        )

        return df

    def __repr__(self) -> str:
        """Simple object representation.

        Returns:
            str: string representation.
        """
        return f"<STAT:{self.id} - HP:{self.hungry_people} - HG:{self.hungry_guests}>"

__table_args__ class-attribute instance-attribute

__table_args__: tuple = PrimaryKeyConstraint(
    "date",
    "guest",
    name="stats_pkey",
    sqlite_on_conflict="REPLACE",
)

Table arguments, used to create primary key (ON CONFLICT options for composite primary key is available only with table_args).

__tablename__ class-attribute instance-attribute

__tablename__: str = 'stats'

Name of the table.

date class-attribute instance-attribute

date: Column = Column(
    Date, nullable=False, server_default=current_date()
)

Day to which the statistics refers to.

guest class-attribute instance-attribute

guest: Column = Column(
    String(20),
    nullable=True,
    default="NotAGuest",
    server_default="NotAGuest",
)

Different kind of guests are identified by the value defined in config files (see config key panel.guest_types). 'NotAGuest' is the value used for locals.

hungry_people class-attribute instance-attribute

hungry_people: Column = Column(
    Integer, nullable=False, default=0, server_default="0"
)

Number of people that ate in a certain day. different kind of guests are identified by the value in guest column.

__repr__

__repr__() -> str

Simple object representation.

Returns:

Type	Description
str	string representation.

Source code in dlunch/models.py

def __repr__(self) -> str:
    """Simple object representation.

    Returns:
        str: string representation.
    """
    return f"<STAT:{self.id} - HP:{self.hungry_people} - HG:{self.hungry_guests}>"

clear classmethod

clear(config: DictConfig) -> int

Clear table and return deleted rows.

Parameters:

Name	Type	Description	Default
config	DictConfig	Hydra configuration dictionary.	required

Returns:

Type	Description
int	deleted rows.

Source code in dlunch/models.py

@classmethod
def clear(self, config: DictConfig) -> int:
    """Clear table and return deleted rows.

    Args:
        config (DictConfig): Hydra configuration dictionary.

    Returns:
        int: deleted rows.
    """

    session = create_session(config)
    with session:
        # Clean menu
        num_rows_deleted = session.execute(delete(self))
        session.commit()
        log.info(
            f"deleted {num_rows_deleted.rowcount} rows from table '{self.__tablename__}'"
        )

    return num_rows_deleted.rowcount

read_as_df classmethod

read_as_df(config: DictConfig, **kwargs) -> DataFrame

Read table as pandas DataFrame.

Parameters:

Name	Type	Description	Default
config	DictConfig	Hydra configuration dictionary.	required

Returns:

Type	Description
DataFrame	dataframe with table content.

Source code in dlunch/models.py

@classmethod
def read_as_df(self, config: DictConfig, **kwargs) -> pd.DataFrame:
    """Read table as pandas DataFrame.

    Args:
        config (DictConfig): Hydra configuration dictionary.

    Returns:
        pd.DataFrame: dataframe with table content.
    """
    df = pd.read_sql_table(
        self.__tablename__,
        create_engine(config=config),
        schema=config.db.get("schema", SCHEMA),
        **kwargs,
    )

    return df

Users

Bases: Data

Table with users that placed an order.

Methods:

Name	Description
__repr__	Simple object representation.
clear	Clear table and return deleted rows.
read_as_df	Read table as pandas DataFrame.

Attributes:

Name	Type	Description
__tablename__	str	Name of the table.
guest	Column	Guest flag (true if guest).
id	Column	User ID.
lunch_time	Column	User selected lunch time.
orders	DeclarativeMeta	Orders connected to each user.
takeaway	Column	Takeaway flag (true if takeaway).

Source code in dlunch/models.py

class Users(Data):
    """Table with users that placed an order."""

    __tablename__: str = "users"
    """Name of the table."""
    id: Column = Column(
        String(100),
        primary_key=True,
        nullable=False,
    )
    """User ID."""
    guest: Column = Column(
        String(20),
        nullable=False,
        default="NotAGuest",
        server_default="NotAGuest",
    )
    """Guest flag (true if guest)."""
    lunch_time: Column = Column(String(7), index=True, nullable=False)
    """User selected lunch time."""
    takeaway: Column = Column(
        Boolean, nullable=False, default=False, server_default=sql_false()
    )
    """Takeaway flag (true if takeaway)."""
    orders: DeclarativeMeta = relationship(
        "Orders",
        back_populates="user_record",
        cascade="all, delete-orphan",
        passive_deletes=True,
    )
    """Orders connected to each user."""

    @classmethod
    def clear(self, config: DictConfig) -> int:
        """Clear table and return deleted rows.

        Args:
            config (DictConfig): Hydra configuration dictionary.

        Returns:
            int: deleted rows.
        """

        session = create_session(config)
        with session:
            # Clean menu
            num_rows_deleted = session.execute(delete(self))
            session.commit()
            log.info(
                f"deleted {num_rows_deleted.rowcount} rows from table '{self.__tablename__}'"
            )

        return num_rows_deleted.rowcount

    @classmethod
    def read_as_df(self, config: DictConfig, **kwargs) -> pd.DataFrame:
        """Read table as pandas DataFrame.

        Args:
            config (DictConfig): Hydra configuration dictionary.

        Returns:
            pd.DataFrame: dataframe with table content.
        """
        df = pd.read_sql_table(
            self.__tablename__,
            create_engine(config=config),
            schema=config.db.get("schema", SCHEMA),
            **kwargs,
        )

        return df

    def __repr__(self) -> str:
        """Simple object representation.

        Returns:
            str: string representation.
        """
        return f"<USER:{self.id}>"

__tablename__ class-attribute instance-attribute

__tablename__: str = 'users'

Name of the table.

guest class-attribute instance-attribute

guest: Column = Column(
    String(20),
    nullable=False,
    default="NotAGuest",
    server_default="NotAGuest",
)

Guest flag (true if guest).

id class-attribute instance-attribute

id: Column = Column(
    String(100), primary_key=True, nullable=False
)

User ID.

lunch_time class-attribute instance-attribute

lunch_time: Column = Column(
    String(7), index=True, nullable=False
)

User selected lunch time.

orders class-attribute instance-attribute

orders: DeclarativeMeta = relationship(
    "Orders",
    back_populates="user_record",
    cascade="all, delete-orphan",
    passive_deletes=True,
)

Orders connected to each user.

takeaway class-attribute instance-attribute

takeaway: Column = Column(
    Boolean,
    nullable=False,
    default=False,
    server_default=false(),
)

Takeaway flag (true if takeaway).

__repr__

__repr__() -> str

Simple object representation.

Returns:

Type	Description
str	string representation.

Source code in dlunch/models.py

def __repr__(self) -> str:
    """Simple object representation.

    Returns:
        str: string representation.
    """
    return f"<USER:{self.id}>"

clear classmethod

clear(config: DictConfig) -> int

Clear table and return deleted rows.

Parameters:

Name	Type	Description	Default
config	DictConfig	Hydra configuration dictionary.	required

Returns:

Type	Description
int	deleted rows.

Source code in dlunch/models.py

@classmethod
def clear(self, config: DictConfig) -> int:
    """Clear table and return deleted rows.

    Args:
        config (DictConfig): Hydra configuration dictionary.

    Returns:
        int: deleted rows.
    """

    session = create_session(config)
    with session:
        # Clean menu
        num_rows_deleted = session.execute(delete(self))
        session.commit()
        log.info(
            f"deleted {num_rows_deleted.rowcount} rows from table '{self.__tablename__}'"
        )

    return num_rows_deleted.rowcount

read_as_df classmethod

read_as_df(config: DictConfig, **kwargs) -> DataFrame

Read table as pandas DataFrame.

Parameters:

Name	Type	Description	Default
config	DictConfig	Hydra configuration dictionary.	required

Returns:

Type	Description
DataFrame	dataframe with table content.

Source code in dlunch/models.py

@classmethod
def read_as_df(self, config: DictConfig, **kwargs) -> pd.DataFrame:
    """Read table as pandas DataFrame.

    Args:
        config (DictConfig): Hydra configuration dictionary.

    Returns:
        pd.DataFrame: dataframe with table content.
    """
    df = pd.read_sql_table(
        self.__tablename__,
        create_engine(config=config),
        schema=config.db.get("schema", SCHEMA),
        **kwargs,
    )

    return df

create_database

create_database(
    config: DictConfig, add_basic_auth_users=False
) -> None

Function to create the database through SQLAlchemy models.

Parameters:

Name	Type	Description	Default
config	DictConfig	Hydra configuration dictionary.	required
add_basic_auth_users	bool	set to true to add admin and guest users. These users are of interest only if 'basic authentication' is selected. Defaults to False.	False

Source code in dlunch/models.py

def create_database(config: DictConfig, add_basic_auth_users=False) -> None:
    """Function to create the database through SQLAlchemy models.

    Args:
        config (DictConfig): Hydra configuration dictionary.
        add_basic_auth_users (bool, optional): set to true to add admin and guest users.
            These users are of interest only if 'basic authentication' is selected.
            Defaults to False.
    """
    # Create directory if missing
    log.debug("create 'shared_data' folder")
    pathlib.Path(config.db.shared_data_folder).mkdir(exist_ok=True)

    # In case the database is not ready use a retry mechanism
    @tenacity.retry(
        retry=tenacity.retry_if_exception_type(OperationalError),
        wait=tenacity.wait_fixed(config.db.create_retries.wait),
        stop=(
            tenacity.stop_after_delay(config.db.create_retries.stop.delay)
            | tenacity.stop_after_attempt(
                config.db.create_retries.stop.attempts
            )
        ),
    )
    def create_database_with_retries(config: DictConfig) -> None:
        engine = create_engine(config)
        Data.metadata.create_all(engine)

    # Create tables
    log.debug(f"attempt database creation: {config.db.attempt_creation}")
    if config.db.attempt_creation:
        create_database_with_retries(config)

        # Retries stats
        log.debug(
            f"create database attempts: {create_database_with_retries.retry.statistics}"
        )

    # If requested add users for basic auth (admin and guest)
    if add_basic_auth_users:
        log.debug("add basic auth standard users")
        # If no user exist create the default admin
        session = create_session(config)

        with session:
            # Check if admin exists
            if session.get(Credentials, "admin") is None:
                # Add authorization and credentials for admin
                auth.add_privileged_user(
                    user="admin",
                    is_admin=True,
                    config=config,
                )
                auth.add_user_hashed_password(
                    user="admin",
                    password="admin",
                    config=config,
                )
            # Check if guest exists
            if (
                session.get(Credentials, "guest") is None
            ) and config.basic_auth.guest_user:
                # Add only credentials for guest (guest users are not included
                # in privileged_users table)
                auth.add_user_hashed_password(
                    user="guest",
                    password="guest",
                    config=config,
                )

create_engine

create_engine(config: DictConfig) -> Engine

Factory function for SQLAlchemy engine.

Parameters:

Name	Type	Description	Default
config	DictConfig	Hydra configuration dictionary.	required

Returns:

Type	Description
Engine	SQLAlchemy engine.

Source code in dlunch/models.py

def create_engine(config: DictConfig) -> Engine:
    """Factory function for SQLAlchemy engine.

    Args:
        config (DictConfig): Hydra configuration dictionary.

    Returns:
        Engine: SQLAlchemy engine.
    """
    engine = hydra.utils.instantiate(config.db.engine)

    # Change schema with change_execution_options
    # If schema exist in config.db it will override the schema selected through
    # the environment variable
    if "schema" in config.db:
        engine.update_execution_options(
            schema_translate_map={SCHEMA: config.db.schema}
        )

    return engine

create_exclusive_session

create_exclusive_session(config: DictConfig) -> Session

Factory function for database exclusive session. Database is locked until the transaction is completed (to be used to avoid race conditions).

Parameters:

Name	Type	Description	Default
config	DictConfig	Hydra configuration dictionary.	required

Returns:

Type	Description
Session	SQLAlchemy exclusive session.

Note

This function is not used inside Data-Lunch code.

Source code in dlunch/models.py

def create_exclusive_session(config: DictConfig) -> Session:
    """Factory function for database exclusive session.
    Database is locked until the transaction is completed (to be used to avoid
    race conditions).

    Args:
        config (DictConfig): Hydra configuration dictionary.

    Returns:
        Session: SQLAlchemy exclusive session.

    Note:
        This function is not used inside Data-Lunch code.

    """
    engine = create_engine(config)

    # Alter begin statement
    @event.listens_for(engine, "connect")
    def do_connect(dbapi_connection, connection_record):
        # disable pysqlite's emitting of the BEGIN statement entirely.
        # also stops it from emitting COMMIT before any DDL.
        dbapi_connection.isolation_level = None

    @event.listens_for(engine, "begin")
    def do_begin(conn):
        # Emit exclusive BEGIN
        conn.exec_driver_sql("BEGIN EXCLUSIVE")

    session = Session(engine)

    return session

create_session

create_session(config: DictConfig) -> Session

Factory function for database session.

Parameters:

Name	Type	Description	Default
config	DictConfig	Hydra configuration dictionary.	required

Returns:

Type	Description
Session	SQLAlchemy session.

Source code in dlunch/models.py

def create_session(config: DictConfig) -> Session:
    """Factory function for database session.

    Args:
        config (DictConfig): Hydra configuration dictionary.

    Returns:
        Session: SQLAlchemy session.
    """
    engine = create_engine(config)
    session = Session(engine)

    return session

get_db_dialect

get_db_dialect(
    db_obj: Session | Connection | Connection,
) -> str

Return database type (postgresql, sqlite, etc.) based on the database object passed as input. If a session is passed, the database type is set based on an internal map (see models._DBTYPE_MAP).

Parameters:

Name	Type	Description	Default
db_obj	Session | Connection | Connection	session or connection object.	required

Raises:

Type	Description
TypeError	db_obj shall be a session or a connection object.

Returns:

Type	Description
str	database dialect.

Source code in dlunch/models.py

def get_db_dialect(
    db_obj: Session | ConnectionSqlite | ConnectionPostgresql,
) -> str:
    """Return database type (postgresql, sqlite, etc.) based on the database object passed as input.
    If a session is passed, the database type is set based on an internal map (see `models._DBTYPE_MAP`).

    Args:
        db_obj (Session | ConnectionSqlite | ConnectionPostgresql): session or connection object.

    Raises:
        TypeError: db_obj shall be a session or a connection object.

    Returns:
        str: database dialect.
    """
    if isinstance(db_obj, Session):
        dialect = db_obj.bind.dialect.name
    elif isinstance(db_obj, ConnectionSqlite) or isinstance(
        db_obj, ConnectionPostgresql
    ):
        module = db_obj.__class__.__module__.split(".", 1)[0]
        dialect = _MODULE_TO_DIALECT_MAP[module]
    else:
        raise TypeError("db_obj should be a session or connection object")

    return dialect

get_flag

get_flag(
    config: DictConfig,
    id: str,
    value_if_missing: bool | None = None,
) -> bool | None

Get the value of a flag. Optionally select the values to return if the flag is missing (default to None).

Parameters:

Name	Type	Description	Default
config	DictConfig	Hydra configuration dictionary.	required
id	str	flag id (name).	required
value_if_missing	bool | None	value to return if the flag does not exist. Defaults to None.	None

Returns:

Type	Description
bool | None	flag value.

Source code in dlunch/models.py

def get_flag(
    config: DictConfig, id: str, value_if_missing: bool | None = None
) -> bool | None:
    """Get the value of a flag.
    Optionally select the values to return if the flag is missing (default to None).

    Args:
        config (DictConfig): Hydra configuration dictionary.
        id (str): flag id (name).
        value_if_missing (bool | None, optional): value to return if the flag does not exist. Defaults to None.

    Returns:
        bool | None: flag value.
    """

    session = create_session(config)
    flag = session.get(Flags, id)
    if flag is None:
        value = value_if_missing
    else:
        value = flag.value
    return value

session_add_with_upsert

session_add_with_upsert(
    session: Session,
    constraint: str,
    new_record: Stats | Flags,
) -> None

Use an upsert statement for postgresql to add a new record to a table, a simple session add otherwise.

Parameters:

Name	Type	Description	Default
session	Session	SQLAlchemy session object.	required
constraint	str	constraint used for upsert (usually the primary key)	required
new_record	Stats | Flags	table resord (valid tables are stats or flags)	required

Source code in dlunch/models.py

def session_add_with_upsert(
    session: Session, constraint: str, new_record: Stats | Flags
) -> None:
    """Use an upsert statement for postgresql to add a new record to a table,
    a simple session add otherwise.

    Args:
        session (Session): SQLAlchemy session object.
        constraint (str): constraint used for upsert (usually the primary key)
        new_record (Stats | Flags): table resord (valid tables are `stats` or `flags`)
    """
    # Use an upsert for postgresql (for sqlite an 'on conflict replace' is set
    # on table, so session.add is fine)
    if get_db_dialect(session) == "postgresql":
        insert_statement = postgresql_upsert(new_record.__table__).values(
            {
                column.name: getattr(new_record, column.name)
                for column in new_record.__table__.c
                if getattr(new_record, column.name, None) is not None
            }
        )
        upsert_statement = insert_statement.on_conflict_do_update(
            constraint=constraint,
            set_={
                column.name: getattr(insert_statement.excluded, column.name)
                for column in insert_statement.excluded
            },
        )
        session.execute(upsert_statement)
    else:
        session.add(new_record)

set_flag

set_flag(config: DictConfig, id: str, value: bool) -> None

Set a key,value pair inside flag table.

Parameters:

Name	Type	Description	Default
config	DictConfig	Hydra configuration dictionary.	required
id	str	flag id (name).	required
value	bool	flag value.	required

Source code in dlunch/models.py

def set_flag(config: DictConfig, id: str, value: bool) -> None:
    """Set a key,value pair inside `flag` table.

    Args:
        config (DictConfig): Hydra configuration dictionary.
        id (str): flag id (name).
        value (bool): flag value.
    """

    session = create_session(config)

    with session:
        # Write the selected flag (it will be overwritten if exists)
        new_flag = Flags(id=id, value=value)

        # Use an upsert for postgresql, a simple session add otherwise
        session_add_with_upsert(
            session=session, constraint="flags_pkey", new_record=new_flag
        )

        session.commit()

set_sqlite_pragma

set_sqlite_pragma(dbapi_connection, connection_record)

Force foreign key constraints for sqlite connections.

Parameters:

Name	Type	Description	Default
dbapi_connection	Any	connection to database. Shall have a cursor method.	required
connection_record	Any	connection record (not used).	required

Source code in dlunch/models.py

@event.listens_for(Engine, "connect")
def set_sqlite_pragma(dbapi_connection, connection_record):
    """Force foreign key constraints for sqlite connections.

    Args:
        dbapi_connection (Any): connection to database. Shall have a `cursor` method.
        connection_record (Any): connection record (not used).
    """
    if get_db_dialect(dbapi_connection) == "sqlite":
        cursor = dbapi_connection.cursor()
        cursor.execute("PRAGMA foreign_keys=ON;")
        cursor.close()