An object-relational mapper (ORM) maps programming-language objects to database records, simplifying interaction between application classes and a relational database. In an ACH-processing application, it can sit between Python code and PostgreSQL so developers work with modeled data rather than writing every database operation directly. section 12.1.2 section 5.1
An ORM is useful, but it is an abstraction rather than a replacement for understanding the database. Relational design still defines field types, table relationships, foreign keys, and cascading deletion. For example, deleting an ACH batch header should also remove its associated entry, addenda, and control records when the schema uses foreign keys with ON DELETE CASCADE. section 5.1 section 12.1.2
A gradual path to ORM adoption can keep database behavior visible:
- Design tables around the data hierarchy and select database types that reflect the stored fields, such as NUMERIC(12,2) for amounts. section 5.1
- Model application data with a Pydantic schema so callers know which fields a record contains. section 5.8
- Wrap handwritten SQL in CRUD methods, then test insertion and retrieval before replacing selected queries with ORM operations. section 5.8 section 12.1.2
From an ACH file object to a stored record
An AchFileSchema holds fields such as file_name, file_hash, and created_at. An AchFileSql method accepts that object, inserts its values into ach_files, returns the generated ID, and retrieves the row as a schema object. This provides an object-shaped interface while the SQL remains explicit; an ORM could later replace some of that handwritten SQL. section 5.8 section 12.1.2
Listing 5.29 AchFileSql class
from typing import Optional #1 from uuid import UUID #1 from psycopg.rows import dict_row #1 from ach_processor.database import DbUtils #1 from ach_processor.schemas.AchFileSchema #1 ➥import AchFileSchema #1 #1 class AchFileSql: def insert_record(self, ach_file: AchFileSchema) ➥ -> UUID: #2 with DbUtils.get_db_connection() as conn: result = conn.execute( """ INSERT INTO ach_files(ach_files_id, ➥ file_name, file_hash, created_at) VALUES ➥(DEFAULT, %(file_name)s, %(file_hash)s, DEFAULT) RETURNING ach_files_id #3 """, ach_file.model_dump()) #4 return result.fetchone()[0] def get_record(self, ach_file_id: UUID) ➥ -> AchFileSchema: #5 with DbUtils.get_db_connection(row_factory=class_row(AchFileSchema)) ➥ as conn: #6 result = conn.execute( """ SELECT * FROM ach_files WHERE ach_files_id = %s """, [ach_file_id.hex]) record = result.fetchone() if not record: #7 raise KeyError(f"Record with id #7 ➥ {ach_file_id} not found") #7 return record
Inspect the CRUD wrapper to see an intermediate design: a typed object is passed into explicit SQL and read back as a schema object. section 5.8
