PostgresEngine

Configuration

# piccolo_conf.py
from piccolo.engine.postgres import PostgresEngine


DB = PostgresEngine(config={
    'host': 'localhost',
    'database': 'my_app',
    'user': 'postgres',
    'password': ''
})

config

The config dictionary is passed directly to the underlying database adapter, asyncpg. See the asyncpg docs to learn more.

---

Connection Pool

See Connection Pool.

---

Source

class piccolo.engine.postgres.PostgresEngine(config: dict[str, Any], extensions: Sequence[str] = (), log_queries: bool = False, log_responses: bool = False, extra_nodes: Mapping[str, PostgresEngine] | None = None)

Used to connect to PostgreSQL.

Parameters:

• config –

  The config dictionary is passed to the underlying database adapter, asyncpg. Common arguments you’re likely to need are:

  • host

  • port

  • user

  • password

  • database

  For example, {'host': 'localhost', 'port': 5432}.

  See the asyncpg docs for all available options.

• extensions – When the engine starts, it will try and create these extensions in Postgres. If you’re using a read only database, set this value to an empty tuple ().

• log_queries – If True, all SQL and DDL statements are printed out before being run. Useful for debugging.

• log_responses – If True, the raw response from each query is printed out. Useful for debugging.

• extra_nodes –

  If you have additional database nodes (e.g. read replicas) for the server, you can specify them here. It’s a mapping of a memorable name to a PostgresEngine instance. For example:

  DB = PostgresEngine(
      config={'database': 'main_db'},
      extra_nodes={
          'read_replica_1': PostgresEngine(
              config={
                  'database': 'main_db',
                  host: 'read_replicate.my_db.com'
              },
              extensions=()
          )
      }
  )
  

  Note how we set extensions=(), because it’s a read only database.

  When executing a query, you can specify one of these nodes instead of the main database. For example:

  >>> await MyTable.select().run(node="read_replica_1")