Skip to content

Postgres

Module to store the PostgreSQL engine.

PostgresqlSqlEngine (SqlEngine)

PostgreSQL engine class.

Source code in now8_api/data/database/postgres.py 
class PostgresqlSqlEngine(SqlEngine):
    """PostgreSQL engine class."""

    @retry(stop=stop_after_attempt(3))
    async def execute_query(self, query: str, *args) -> List:
        """Return the result of executing the passed query.

        Arguments:
            query: Query to perform.

        Returns:
            The result of executing the passed query.
        """
        db_conf = self.get_db_conf()
        db_conf["database"] = db_conf.pop("dbname")

        try:
            conn = await asyncpg.connect(**db_conf, timeout=3)
        except socket.gaierror as error:
            raise ConnectionError(
                f"Can't connect to database: {db_conf}."
            ) from error

        query_result = await conn.fetch(query, *args)

        await conn.close()

        return query_result

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

Returns:

Type Description
List

The result of executing the passed query.
Source code in now8_api/data/database/postgres.py 
@retry(stop=stop_after_attempt(3))
async def execute_query(self, query: str, *args) -> List:
    """Return the result of executing the passed query.

    Arguments:
        query: Query to perform.

    Returns:
        The result of executing the passed query.
    """
    db_conf = self.get_db_conf()
    db_conf["database"] = db_conf.pop("dbname")

    try:
        conn = await asyncpg.connect(**db_conf, timeout=3)
    except socket.gaierror as error:
        raise ConnectionError(
            f"Can't connect to database: {db_conf}."
        ) from error

    query_result = await conn.fetch(query, *args)

    await conn.close()

    return query_result
Last update: 2022-08-05