A database connection is the session an application uses to communicate with a database server. In an ACH-processing workflow, it turns parsed file data into stored records that a later dashboard can aggregate. Establish the connection before parsing or inserting records, then use it to obtain a cursor or execute SQL directly. section 5.2.1 section 5.3.2
Start by proving the basics: create a connection, obtain a cursor, and close both resources. This small check separates database-access problems from parsing problems before inserts are added. A connection string must point to a reachable server; when code runs from the machine hosting Docker, use localhost rather than the Docker service name postgres. section 5.2.1
Listing 5.4 Testing the connection
def parse(self, filename) -> [List, List]:
conn = self.get_db() #1
cursor = conn.cursor() #2
cursor.close() #3
conn.close() #4
with open(filename, "r") as file:
lines = file.readlines()
This minimal routine isolates the connection lifecycle: get the connection, create a cursor for SQL work, then close the cursor and connection. section 5.2.1
Container networking is part of making a local connection work. Expose PostgreSQL port 5432 in docker-compose so the host-based program can reach the container. An unknown-host error points to the wrong hostname; a socket or port failure suggests that the port is unavailable or not exposed. Retrace the setup when these errors occur rather than assuming the parsing code is at fault. section 5.2.1
Once connectivity works, use the connection deliberately:
- Open it around the file-processing work with a context-managed with statement instead of relying on manual closing. section 5.2.1
- Create a cursor within the connection when executing row inserts, then commit after the insert operation. section 5.2.1
- Pass an existing connection into a parsing routine when that routine stores parsed data; this makes the routine easier to unit test than storing the connection on the class. section 5.3.2
- Watch for duplicated connection helpers and consider extracting them into utility or helper classes as the application grows. section 5.3.2
Listing 5.6 Updated parse function
def parse(self, filename) -> [List, List]:
with open(filename, "r") as file, self.get_db() as conn: #1
lines = file.readlines()
sequence_number = 0 #2
for line in lines:
sequence_number += 1 #3
line = line.replace("\n", "")
with conn.cursor() as cursor: #4
cursor.execute(f"INSERT INTO ach_files #4
➥ (file_name, unparsed_record, sequence_number) #4
➥ VALUES (%s, %s, %s)", (filename, line, sequence_number)) #4
conn.commit() #4
The parser keeps the database connection open while reading ACH lines, opens a cursor for each insert, and commits the stored record. section 5.2.1
Persisting an ACH file header
After parsing a header line into named fields, pass the connection into the header routine and execute an INSERT into ach_file_headers. A unit test can open its own connection, run the parser, and truncate the table during teardown so stored rows from one test do not affect the next. section 5.3.2 section 5.2.1
