Python for DevOps: Working with Databases and Persistent Storage

Modern DevOps automation does not just move files and call APIs — it reads and writes data. Whether you are syncing configuration, backfilling metrics, or migrating records between environments, your scripts need reliable access to databases and caches. In this lesson you will connect to PostgreSQL with psycopg, use Redis with redis-py, manage connection pools, handle transaction failures gracefully, and build a migration script that logs every step.

1. Learning Objectives

By the end of this lesson, you will be able to:

• Connect to PostgreSQL from Python using psycopg connection strings
• Use the redis-py client to read and write cached values
• Bound database connections with ConnectionPool so scripts never leak connections
• Write transactions that commit only when every statement succeeds
• Handle database errors such as UniqueViolation and aborted transactions
• Build an idempotent data migration script that logs every step

2. Why This Matters (Real-World Scenario)

Imagine a Friday release: your team moved the inventory table into a new schema, but the old cron job still reads the legacy table. Someone runs the sync script twice and it crashes halfway because of duplicate keys, leaving the database in a half-migrated state. Without connection limits, the same script exhausts PostgreSQL connections and takes down the application with it. This is exactly the incident that database-aware Python prevents: bounded connections, safe transactions, idempotent writes, and a migration that logs every step so you can see precisely where it stopped.

3. Core Concepts

Relational databases (PostgreSQL) store structured rows with constraints, joins, and ACID transactions. Use them for anything that must never lose data: user records, deployments, inventory. Key-value stores (Redis) keep data in memory for speed: caches, rate-limit counters, feature flags, and short-lived state. A common DevOps pattern is PostgreSQL as the source of truth and Redis as the fast read layer in front of it.

psycopg is the standard PostgreSQL adapter for Python. Version 3 (installed as psycopg[binary]) is the modern default; you will still see psycopg2 in older scripts. The connection string is a URL: postgresql://user:password@host:port/dbname. Always pass values as %s placeholders - never by string formatting - to avoid SQL injection and quoting bugs.

redis-py is the official Redis client. With decode_responses=True it returns strings instead of bytes, which keeps your scripts readable. Values can be given a TTL (time to live) so stale cache entries expire automatically.

Connection pools reuse a fixed set of connections instead of opening a new one per task. Opening a connection takes tens of milliseconds and every open connection consumes memory on the server, so unbounded connections are a classic production incident.

Transactions group statements so they either all commit or all roll back. In psycopg, entering the connection context manager starts a transaction and exiting it cleanly commits; any exception rolls back instead.

4. Hands-On Practice: Building a Data Sync Pipeline

We will build the pieces of a real DevOps data pipeline: an environment, a PostgreSQL connection, a bounded pool, a safe transaction, a Redis cache, and finally one migration script that ties them together.

Step 1: Set up the environment. Create a virtual environment and install the libraries:

python3 -m venv venv
source venv/bin/activate
pip install 'psycopg[binary]' psycopg-pool redis

Step 2: Connect to PostgreSQL. Keep the connection string in an environment variable such as DATABASE_URL, then connect:

import os
import psycopg

DATABASE_URL = os.getenv('DATABASE_URL', 'postgresql://admin:REDACTED@localhost:5432/devops')

with psycopg.connect(DATABASE_URL) as conn:
    with conn.cursor() as cur:
        cur.execute('SELECT version();')
        print(cur.fetchone()[0])

Step 3: Bound connections with a pool. Opening a new connection per task is slow and can exhaust the server. Use ConnectionPool from psycopg_pool:

from psycopg_pool import ConnectionPool

pool = ConnectionPool(
    'postgresql://admin:REDACTED@localhost:5432/devops',
    min_size=1,
    max_size=10,
)

with pool.connection() as conn:
    with conn.cursor() as cur:
        cur.execute('SELECT count(*) FROM servers;')
        print('servers:', cur.fetchone()[0])

pool.close()

Step 4: Make transactions safe. psycopg commits when the with block exits cleanly and rolls back on any exception. Never leave a multi-statement operation half-applied:

import psycopg

def record_deployment(conn, server_id, status):
    with conn.cursor() as cur:
        cur.execute(
            'INSERT INTO deployments (server_id, status, deployed_at) '
            'VALUES (%s, %s, now());',
            (server_id, status),
        )

try:
    with psycopg.connect('postgresql://admin:REDACTED@localhost:5432/devops') as conn:
        record_deployment(conn, 42, 'ok')
        # commit happens automatically when the with block exits cleanly
except psycopg.Error as exc:
    print('rolled back:', exc)

Step 5: Cache with Redis. For values that are expensive to compute, cache them with a TTL so they expire automatically:

import redis

r = redis.Redis(host='localhost', port=6379, db=0, decode_responses=True)
r.set('deploy:latest', 'release-1.4.2', ex=600)
print('cached value:', r.get('deploy:latest'))
print('TTL seconds:', r.ttl('deploy:latest'))

Step 6: Build the logged migration script. Now combine everything into one idempotent migration that logs every step:

#!/usr/bin/env python3
# Migrate legacy server records into the new schema, with logging.
import logging
import psycopg
from psycopg_pool import ConnectionPool

logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s %(levelname)s %(name)s: %(message)s',
)
log = logging.getLogger('migrate')

POOL = ConnectionPool(
    'postgresql://admin:REDACTED@localhost:5432/devops',
    min_size=1,
    max_size=5,
)


def migrate_servers():
    processed = 0
    with POOL.connection() as conn:
        with conn.cursor() as cur:
            cur.execute('SELECT id, hostname, ip FROM servers_legacy;')
            rows = cur.fetchall()
        log.info('read %d rows from servers_legacy', len(rows))
        for row in rows:
            with conn.cursor() as cur:
                cur.execute(
                    'INSERT INTO servers (hostname, ip) VALUES (%s, %s) '
                    'ON CONFLICT (hostname) DO NOTHING;',
                    (row[1], row[2]),
                )
            processed += 1
            log.debug('migrated %s', row[1])
    log.info('migration complete: %d servers processed', processed)


if __name__ == '__main__':
    migrate_servers()
    POOL.close()

Run it and you should see output like this:

Run it a second time - the ON CONFLICT DO NOTHING clause turns it into a no-op instead of a duplicate-key failure. That is what makes a migration safe to re-run.

5. Common Errors & Solutions

1. psycopg.OperationalError: connection refused - PostgreSQL is not running or the port is wrong. Check with pg_isready and systemctl status postgresql, then verify the host and port in your connection string.

2. password authentication failed for user - The credentials in the connection string are wrong, or pg_hba.conf rejects the connection. Keep credentials in environment variables, never hardcode them, and confirm the role has a password set.

3. UniqueViolation: duplicate key value violates unique constraint - A re-run tried to insert a row that already exists. Make the script idempotent with ON CONFLICT (column) DO NOTHING or DO UPDATE so it is safe to execute twice.

4. InFailedSqlTransaction: current transaction is aborted - After one statement fails, PostgreSQL refuses further commands in the same transaction. Catch the exception and roll back (or exit the with block) before retrying, or use savepoints for partial rollback.

5. redis.exceptions.ConnectionError: Error 111 connecting to localhost:6379. Connection refused. - Redis is not running. Start the service and verify with redis-cli ping, which should reply PONG.

6. Summary Checklist

Before you move on, confirm each item:

• PostgreSQL is reachable with pg_isready
• psycopg connects using credentials from environment variables
• ConnectionPool bounds connections with min_size and max_size
• Transactions commit only after every statement succeeds
• Redis responds to redis-cli ping with PONG
• Your migration script is idempotent and safe to re-run
• Every phase of the migration is logged

7. Practice Exercise

Build a sync_inventory.py script that: reads a JSON file of servers, upserts each record into PostgreSQL with ON CONFLICT DO UPDATE, caches the result in Redis with a 10-minute TTL, and logs each phase with the logging module. Run it twice and confirm the second run creates no duplicates and does not fail.

8. Next Steps

You now have database-aware Python: bounded connection pools, safe transactions, Redis caching, and idempotent, logged migrations. In the next lesson, we will cover Python Testing with pytest: Writing Reliable Automation Tests - writing unit and integration tests for your scripts, mocking database calls, and running tests in CI so your automation breaks loudly in a pipeline instead of silently in production.