Python Enterprise Connectors

Python Enterprise Connectors

Overview

Enterprise database and mainframe connectivity in Python requires specific drivers, careful credential management, and understanding of legacy data formats. This skill covers DB2, Oracle, SQL Server, mainframe systems, and patterns for building robust connectors with proper security.

IBM DB2

Drivers

Connection Patterns

import ibm_db

# DB2 LUW (Linux/Unix/Windows)
conn = ibm_db.connect(
    f"DATABASE={db};HOSTNAME={host};PORT={port};PROTOCOL=TCPIP;UID={user};PWD={pwd};SECURITY=SSL;",
    "", ""
)

# DB2 z/OS (mainframe) — REQUIRES DB2 Connect license
conn = ibm_db.connect(
    f"DATABASE={location};HOSTNAME={host};PORT={port};PROTOCOL=TCPIP;UID={user};PWD={pwd};",
    "", ""
)

Critical gotcha: DB2 z/OS connections require a paid DB2 Connect license. Without it: SQL1598N error. This is the #1 issue developers hit.

Performance Tuning

# Increase fetch size for large result sets
stmt = ibm_db.exec_immediate(conn, sql)
ibm_db.set_option(stmt, {ibm_db.SQL_ATTR_CURSOR_TYPE: ibm_db.SQL_CURSOR_FORWARD_ONLY}, 1)
# Use array fetch for bulk operations

EBCDIC Handling

Mainframe DB2 (z/OS) uses EBCDIC encoding. Known issues with ibm_db error message decoding.

# Convert EBCDIC to UTF-8
data.decode('cp500').encode('utf-8')  # CP500 = common EBCDIC code page
# Or use codecs
import codecs
text = codecs.decode(ebcdic_bytes, 'cp037')  # CP037 = US EBCDIC

Common EBCDIC code pages: CP037 (US/Canada), CP500 (International), CP1047 (z/OS Unix).

Mainframe Connectivity

Zowe Python SDK (Modern Approach)

pip install zowe-zos-files-for-zowe-sdk zowe-zos-jobs-for-zowe-sdk zowe-zos-tso-for-zowe-sdk

Status (2026): Pre-release (v1.0.0.dev25) but actively developed. For production, consider Zowe CLI with subprocess calls as a more stable alternative.

z/OSMF REST API (Direct)

import httpx

client = httpx.Client(base_url="https://mainframe:443/zosmf", verify=True,
                      auth=(user, password))

# List datasets
response = client.get("/restfiles/ds", params={"dslevel": "USER.PROD.*"})

# Submit JCL
response = client.put("/restjobs/jobs",
    content=jcl_content,
    headers={"Content-Type": "text/plain"})
job_id = response.json()["jobid"]

# Check job status
response = client.get(f"/restjobs/jobs/{job_name}/{job_id}")

Terminal Emulation (3270)

IBM MQ (Messaging)

# Package renamed: pymqi → ibmmq (v2.0.3, Feb 2026)
import ibmmq

qmgr = ibmmq.connect(queue_manager)
queue = ibmmq.Queue(qmgr, 'QUEUE.NAME')
queue.put('message data')
message = queue.get()
queue.close()
qmgr.disconnect()

z/OS Connect (REST → CICS/IMS)

z/OS Connect 3.0 provides RESTful access to CICS transactions and IMS programs. This is the recommended path for Python-to-CICS/IMS since CTG has no native Python client.

# Call a CICS transaction via z/OS Connect
response = httpx.post(
    "https://zosconnect:9443/zosConnect/apis/myApi/operation",
    json={"input_field": "value"},
    auth=(user, password)
)

Other Enterprise Databases

Deprecated — Migrate Away

Connector Building Patterns

DB-API 2.0 Compliance

All Python database drivers should implement PEP 249:

• connect() → connection object
• connection.cursor() → cursor object
• cursor.execute(sql, params) → parameterized queries
• cursor.fetchone(), fetchmany(), fetchall()
• connection.commit(), connection.rollback()

Connection Factory

class DatabaseConnector:
    def __init__(self, config: dict):
        self.config = config
        self._engine = None

    def get_engine(self):
        if self._engine is None:
            self._engine = create_engine(
                self.config['url'],
                pool_size=self.config.get('pool_size', 10),
                pool_pre_ping=True,
                pool_recycle=300,
            )
        return self._engine

    def health_check(self) -> bool:
        try:
            with self.get_engine().connect() as conn:
                conn.execute(text("SELECT 1"))
            return True
        except Exception:
            return False

Retry Pattern

from tenacity import retry, stop_after_attempt, wait_exponential, retry_if_exception_type

@retry(
    stop=stop_after_attempt(3),
    wait=wait_exponential(multiplier=1, min=2, max=30),
    retry=retry_if_exception_type((ConnectionError, TimeoutError)),
)
def execute_query(engine, sql, params):
    with engine.connect() as conn:
        return conn.execute(text(sql), params)

Credential Security

Hierarchy (Best → Worst)

1. HashiCorp Vault dynamic secrets (auto-rotating, short-lived)
2. Cloud secret managers (AWS Secrets Manager, Azure Key Vault, GCP Secret Manager)
3. Environment variables (acceptable for simple deployments)
4. Config files (encrypted, not in git)
5. Hardcoded in code — NEVER

# Vault dynamic secrets (hvac library)
import hvac
client = hvac.Client(url='https://vault:8200', token=os.environ['VAULT_TOKEN'])
creds = client.secrets.database.generate_credentials('my-role')
username, password = creds['data']['username'], creds['data']['password']
# Credentials auto-expire after TTL

Connection String Safety

# NEVER log connection strings with passwords
import re
def sanitize_url(url: str) -> str:
    return re.sub(r'://[^:]+:[^@]+@', '://***:***@', url)

Data Type Gotchas

Anti-Patterns