init

Module to store the common objects for the database.

SqlEngine (ABC) dataclass

SQL engine class.

Attributes:

Name	Type	Description
env_prefix	str	Prefix of the environment variables with the database connection parameters.
name	Optional[str]	Name of the database.
user	Optional[str]	User name of the database.
password	Optional[str]	Password of the database.
host	Optional[str]	Host of the database.
port	Optional[str]	Port of the database.

Source code in now8_api/data/database/__init__.py

@dataclass
class SqlEngine(ABC):
    """SQL engine class.

    Attributes:
        env_prefix: Prefix of the environment variables with the database
            connection parameters.
        name: Name of the database.
        user: User name of the database.
        password: Password of the database.
        host: Host of the database.
        port: Port of the database.
    """

    env_prefix: str = "DB_"
    name: Optional[str] = None
    user: Optional[str] = None
    password: Optional[str] = None
    host: Optional[str] = None
    port: Optional[str] = None

    @validator("name", always=True)
    @classmethod
    def validate_name(cls, value, values):
        """Get name from environment variable if undefined."""
        return (
            environ.get(values["env_prefix"] + "NAME", "postgres")
            if value is None
            else value
        )

    @validator("user", always=True)
    @classmethod
    def validate_user(cls, value, values):
        """Get user from environment variable if undefined."""
        return (
            environ.get(values["env_prefix"] + "USER", "postgres")
            if value is None
            else value
        )

    @validator("password", always=True)
    @classmethod
    def validate_password(cls, value, values):
        """Get password from environment variable if undefined."""
        return (
            environ.get(values["env_prefix"] + "PASS", "postgres")
            if value is None
            else value
        )

    @validator("host", always=True)
    @classmethod
    def validate_host(cls, value, values):
        """Get host from environment variable if undefined."""
        return (
            environ.get(values["env_prefix"] + "HOST", "postgres")
            if value is None
            else value
        )

    @validator("port", always=True)
    @classmethod
    def validate_port(cls, value, values):
        """Get port from environment variable if undefined."""
        return (
            environ.get(values["env_prefix"] + "PORT", "5432")
            if value is None
            else value
        )

    def get_db_conf(self) -> Dict[str, str]:
        """Return the database connection parameters.

        Returns:
            Dictionary with the available database connection parameters.
        """
        return {
            "dbname": str(self.name),
            "user": str(self.user),
            "password": str(self.password),
            "host": str(self.host),
            "port": str(self.port),
        }

    @abstractmethod
    async def execute_query(self, query: str, *args) -> List[tuple]:
        """Return the result of executing the passed query.

        Arguments:
            query: Query to perform.
            args: Query arguments.

        Returns:
            The result of executing the passed query.
        """

__pydantic_model__ (BaseModel) pydantic-model

Source code in now8_api/data/database/__init__.py

@dataclass
class SqlEngine(ABC):
    """SQL engine class.

    Attributes:
        env_prefix: Prefix of the environment variables with the database
            connection parameters.
        name: Name of the database.
        user: User name of the database.
        password: Password of the database.
        host: Host of the database.
        port: Port of the database.
    """

    env_prefix: str = "DB_"
    name: Optional[str] = None
    user: Optional[str] = None
    password: Optional[str] = None
    host: Optional[str] = None
    port: Optional[str] = None

    @validator("name", always=True)
    @classmethod
    def validate_name(cls, value, values):
        """Get name from environment variable if undefined."""
        return (
            environ.get(values["env_prefix"] + "NAME", "postgres")
            if value is None
            else value
        )

    @validator("user", always=True)
    @classmethod
    def validate_user(cls, value, values):
        """Get user from environment variable if undefined."""
        return (
            environ.get(values["env_prefix"] + "USER", "postgres")
            if value is None
            else value
        )

    @validator("password", always=True)
    @classmethod
    def validate_password(cls, value, values):
        """Get password from environment variable if undefined."""
        return (
            environ.get(values["env_prefix"] + "PASS", "postgres")
            if value is None
            else value
        )

    @validator("host", always=True)
    @classmethod
    def validate_host(cls, value, values):
        """Get host from environment variable if undefined."""
        return (
            environ.get(values["env_prefix"] + "HOST", "postgres")
            if value is None
            else value
        )

    @validator("port", always=True)
    @classmethod
    def validate_port(cls, value, values):
        """Get port from environment variable if undefined."""
        return (
            environ.get(values["env_prefix"] + "PORT", "5432")
            if value is None
            else value
        )

    def get_db_conf(self) -> Dict[str, str]:
        """Return the database connection parameters.

        Returns:
            Dictionary with the available database connection parameters.
        """
        return {
            "dbname": str(self.name),
            "user": str(self.user),
            "password": str(self.password),
            "host": str(self.host),
            "port": str(self.port),
        }

    @abstractmethod
    async def execute_query(self, query: str, *args) -> List[tuple]:
        """Return the result of executing the passed query.

        Arguments:
            query: Query to perform.
            args: Query arguments.

        Returns:
            The result of executing the passed query.
        """

validate_host(value, values) classmethod

Get host from environment variable if undefined.

Source code in now8_api/data/database/__init__.py

@validator("host", always=True)
@classmethod
def validate_host(cls, value, values):
    """Get host from environment variable if undefined."""
    return (
        environ.get(values["env_prefix"] + "HOST", "postgres")
        if value is None
        else value
    )

validate_name(value, values) classmethod

Get name from environment variable if undefined.

Source code in now8_api/data/database/__init__.py

@validator("name", always=True)
@classmethod
def validate_name(cls, value, values):
    """Get name from environment variable if undefined."""
    return (
        environ.get(values["env_prefix"] + "NAME", "postgres")
        if value is None
        else value
    )

validate_password(value, values) classmethod

Get password from environment variable if undefined.

Source code in now8_api/data/database/__init__.py

@validator("password", always=True)
@classmethod
def validate_password(cls, value, values):
    """Get password from environment variable if undefined."""
    return (
        environ.get(values["env_prefix"] + "PASS", "postgres")
        if value is None
        else value
    )

validate_port(value, values) classmethod

Get port from environment variable if undefined.

Source code in now8_api/data/database/__init__.py

@validator("port", always=True)
@classmethod
def validate_port(cls, value, values):
    """Get port from environment variable if undefined."""
    return (
        environ.get(values["env_prefix"] + "PORT", "5432")
        if value is None
        else value
    )

validate_user(value, values) classmethod

Get user from environment variable if undefined.

Source code in now8_api/data/database/__init__.py

@validator("user", always=True)
@classmethod
def validate_user(cls, value, values):
    """Get user from environment variable if undefined."""
    return (
        environ.get(values["env_prefix"] + "USER", "postgres")
        if value is None
        else value
    )

execute_query(self, query, *args) async

Return the result of executing the passed query.

Parameters:

Name	Type	Description	Default
query	str	Query to perform.	required
args		Query arguments.	()

Returns:

Type	Description
List[tuple]	The result of executing the passed query.

Source code in now8_api/data/database/__init__.py

@abstractmethod
async def execute_query(self, query: str, *args) -> List[tuple]:
    """Return the result of executing the passed query.

    Arguments:
        query: Query to perform.
        args: Query arguments.

    Returns:
        The result of executing the passed query.
    """

get_db_conf(self)

Return the database connection parameters.

Returns:

Type	Description
Dict[str, str]	Dictionary with the available database connection parameters.

Source code in now8_api/data/database/__init__.py

def get_db_conf(self) -> Dict[str, str]:
    """Return the database connection parameters.

    Returns:
        Dictionary with the available database connection parameters.
    """
    return {
        "dbname": str(self.name),
        "user": str(self.user),
        "password": str(self.password),
        "host": str(self.host),
        "port": str(self.port),
    }

validate_host(value, values) classmethod

Get host from environment variable if undefined.

Source code in now8_api/data/database/__init__.py

@validator("host", always=True)
@classmethod
def validate_host(cls, value, values):
    """Get host from environment variable if undefined."""
    return (
        environ.get(values["env_prefix"] + "HOST", "postgres")
        if value is None
        else value
    )

validate_name(value, values) classmethod

Get name from environment variable if undefined.

Source code in now8_api/data/database/__init__.py

@validator("name", always=True)
@classmethod
def validate_name(cls, value, values):
    """Get name from environment variable if undefined."""
    return (
        environ.get(values["env_prefix"] + "NAME", "postgres")
        if value is None
        else value
    )

validate_password(value, values) classmethod

Get password from environment variable if undefined.

Source code in now8_api/data/database/__init__.py

@validator("password", always=True)
@classmethod
def validate_password(cls, value, values):
    """Get password from environment variable if undefined."""
    return (
        environ.get(values["env_prefix"] + "PASS", "postgres")
        if value is None
        else value
    )

validate_port(value, values) classmethod

Get port from environment variable if undefined.

Source code in now8_api/data/database/__init__.py

@validator("port", always=True)
@classmethod
def validate_port(cls, value, values):
    """Get port from environment variable if undefined."""
    return (
        environ.get(values["env_prefix"] + "PORT", "5432")
        if value is None
        else value
    )

validate_user(value, values) classmethod

Get user from environment variable if undefined.

Source code in now8_api/data/database/__init__.py

@validator("user", always=True)
@classmethod
def validate_user(cls, value, values):
    """Get user from environment variable if undefined."""
    return (
        environ.get(values["env_prefix"] + "USER", "postgres")
        if value is None
        else value
    )

---

Last update: 2022-08-05