QPooler

Centralized PostgreSQL Connection Pool for QGIS Plugins

QGIS 3.28+ | Python 3.9+
View on GitHub

The Problem

Without QPooler, each QGIS plugin creates its own PostgreSQL connections independently. This leads to resource exhaustion, poor performance, and server overload.

Without QPooler - Connection Chaos QGIS Application Plugin A Plugin B Plugin C 5 connections 5 connections 5 connections PostgreSQL Server 15+ Connections! Server Overloaded 50-200ms latency per query

Without QPooler

  • Total Connections15+
  • Latency per Query50-200ms
  • Cache Hit Rate0%
  • Server ProtectionNone
  • MetricsNone

With QPooler

  • Total Connections3-4
  • Latency per Query~0ms (reuse)
  • Cache Hit Rate84%
  • Server ProtectionFull
  • MetricsPer Plugin

The Solution

QPooler acts as a centralized connection broker. All plugins share a single optimized pool, dramatically reducing resource usage and improving performance.

With QPooler - Optimized Connection Pool QGIS Plugins Plugin A Plugin B Plugin C QPooler Pool Manager Cache (84% hit) Policy Engine Metrics Transactions SQL Helpers 3-4 conn PostgreSQL

Architecture Overview

QPooler follows a modular architecture with clear separation of concerns. Each component has a single responsibility.

QPooler Architecture PUBLIC API LAYER db.configure() db.fetch_all() / fetch_one() db.execute() db.transaction() CORE SERVICES Policy Engine Role validation Table whitelist/blacklist Rate limiting Pool Manager Connection pooling Acquire / Release Health checks Cache Manager TTL-based caching LRU eviction LISTEN/NOTIFY Metrics Collector Query statistics Performance tracking Error counting CONNECTION LAYER Wrapper 1 Wrapper 2 Wrapper N psycopg2 UTILITIES SQL Helper Spatial Config WATCHDOG THREAD Health checks | Cleanup idle | Timeouts Transaction timeout | Server lifetime PostgreSQL Server PostGIS | LISTEN/NOTIFY
core/api.py

Public facade - single entry point for all database operations

pool/manager.py

Connection pooling with singleton pattern and watchdog thread

cache/manager.py

TTL-based query caching with LRU eviction

policy/engine.py

Access control with role-based permissions

metrics/collector.py

Performance tracking per plugin

transaction/context.py

Transaction management with savepoints

Request Flow

Every database request follows a consistent path through QPooler's components.

Plugin Request db.fetch_all() Policy.validate() Check permissions Cache? HIT Return Cached MISS Pool.acquire() Get connection Conn.execute() Run SQL Cache.set() Pool.release() Metrics.record()
1
API Call - Plugin calls db.fetch_all() with SQL and parameters
2
Policy Validation - Check role permissions, table access, rate limits
3
Cache Lookup - If caching enabled, check for cached result (84% hit rate)
4
Connection Acquire - Get existing or create new connection from pool
5
Execute Query - Run SQL via psycopg2 connection
6
Finalize - Store in cache, release connection, record metrics

Key Features

Connection Pooling

1 connection per plugin, automatic reuse. Reduces 15+ connections to 3-4.

Query Caching

TTL-based caching with 84% hit rate. LISTEN/NOTIFY for automatic invalidation.

Policy Engine

Role-based access control (read/write/admin). Table whitelist/blacklist. Rate limiting.

Metrics Collection

Per-plugin statistics: query count, avg time, percentiles (p50, p90, p99).

Transaction Support

Context manager with auto-commit/rollback. Savepoint support.

SQL Helpers

Safe query builders. PostGIS helpers with explicit SRID.

Watchdog Thread

Independent health checks, idle cleanup, transaction timeouts.

QGIS Integration

Uses native QGIS connections. Proper lifecycle management.

Policy Roles

Role SELECT INSERT UPDATE DELETE DDL
read Yes No No No No
write Yes Yes Yes Yes No
admin Yes Yes Yes Yes Yes

Quick Start

Basic Usage

from qpooler import db

# Configure from existing QGIS connection
db.configure_from_qgis("my_connection")
db.set_plugin_id("my_plugin")

# Execute queries
rows = db.fetch_all("SELECT * FROM parcels WHERE active = %s", (True,))
row = db.fetch_one("SELECT COUNT(*) FROM parcels")

# With caching (60s TTL)
rows = db.fetch_all("SELECT * FROM ref_table", use_cache=True, cache_ttl=60)

Transactions

with db.transaction(role="write") as tx:
    tx.execute("INSERT INTO logs (msg) VALUES (%s)", ("hello",))
    tx.execute("UPDATE counters SET val = val + 1")
    # Auto-commit on success, auto-rollback on exception

SQL Helpers

from qpooler.sql import SQLHelper, SpatialHelper, QueryBuilder

# Simple SELECT
sql, params = SQLHelper.select("users", ["id", "name"], "active = %s", (True,))

# PostGIS with explicit SRID
sql, params = SpatialHelper.st_intersects(
    "parcels", "geom", "POINT(0 0)", srid=2154
)

# Fluent query builder
sql, params = (QueryBuilder()
    .select("id", "name")
    .from_table("users")
    .where("active = %s", (True,))
    .order_by("name")
    .limit(10)
    .build())

Diagnostics

# Full status
status = db.status()
# {"pool": {...}, "cache": {...}, "metrics": {...}}

# Test connection
ok = db.test_connection()

# Get metrics
metrics = db.get_metrics()

# Check policy violations
violations = db.get_violations()

API Reference

Configuration

Method Description
db.configure(dsn, ...) Configure pool with DSN string
db.configure_from_qgis(name) Configure from QGIS connection
db.set_plugin_id(id) Set default plugin identifier
db.list_qgis_connections() List available QGIS connections

Query Execution

Method Description
db.fetch_all(sql, params, ...) Execute SELECT, return all rows
db.fetch_one(sql, params, ...) Execute SELECT, return single row
db.execute(sql, params, ...) Execute INSERT/UPDATE/DELETE
db.transaction(role) Context manager for transactions

Diagnostics

Method Description
db.status() Full diagnostic status
db.test_connection() Test database connectivity
db.get_metrics() Get performance metrics
db.get_violations() Get policy violations