Database

Module: thinkiq.database

Summary

  • Single class per file: this file defines Database only

  • No global functions/state: everything is instance-level

  • Logger acquired in __init__ (no logging at import time)

  • Centralized helpers: query_one/query_all/execute_non_query

  • TTL cache for SELECT, invalidated on any write

  • Health-check + auto-reconnect

Database

class thinkiq.database.Database

Bases: object

Single access point to Postgres:

  • config/connection/reconnect

  • centralized queries and cache

  • profiling and logging

cache = <thinkiq.ttl_cache.TTLCache object>
static connect(username=None, password=None, host=None, port=None, database=None, url=None)

Creates/re-creates connection. Credentials via thinkiq.config + Crypto. autocommit=True. Lazy-init of schema.pgsql_schema.

Return type:

RealDictConnection

static cursor(profile_tag=None)

Provides a cursor with basic time profiling.

static disconnect()

Safely close connection.

Return type:

None

static execute_non_query(sql, params=None, *, profile_tag=None)

INSERT/UPDATE/DELETE/DDL. Returns rowcount and invalidates cache.

Return type:

int

static get_connection()

Return live connection. Reconnect + health-check if needed.

Return type:

RealDictConnection

static get_connection_fast()

Return live connection without extra health-check; fallback to get_connection() if missing.

Return type:

RealDictConnection

logger = <Logger thinkiq.database (INFO)>
persist_connection: bool = False
static query_all(sql, params=None, *, profile_tag=None, cache_ttl_seconds=0.0)

SELECT returning a list of dict rows. Optional TTL cache.

Return type:

List[Dict[str, Any]]

static query_one(sql, params=None, *, profile_tag=None, cache_ttl_seconds=0.0)

SELECT returning a single dict row or None. Optional TTL cache.

Return type:

Optional[Dict[str, Any]]

static set_persist_connection(enabled=True)

Toggle persisted-connection mode.

Return type:

None

Description

The Database class provides a single access point to PostgreSQL:

  • Connection management - Persistent or per-call connection modes - Health-check and auto-reconnect

  • Centralized query helpers - query_one – fetch a single row as a dict - query_all – fetch multiple rows as a list of dicts - execute_non_query – execute INSERT/UPDATE/DELETE/DDL

  • Caching - Built-in TTL cache for SELECT queries - Automatic invalidation on writes

  • Profiling and logging - Logs queries and execution time with optional tags

Example

from thinkiq.database import Database

# Create a database access object
db = Database(persist_connection=True)

# Get a connection (auto-reconnect enabled)
conn = db.get_connection()

# Run queries
row = db.query_one("SELECT * FROM users WHERE id=%s", (123,))
rows = db.query_all("SELECT * FROM users")

# Insert/update
db.execute_non_query("UPDATE users SET name=%s WHERE id=%s", ("Alice", 123))

# Disconnect safely
db.disconnect()