PostgreSQL + Python for Beginners 🐘🐍 (Part 2)

pip install dbfread psycopg2-binary

import os
from dbfread import DBF, FieldParser
import psycopg2
from psycopg2.errors import UndefinedColumn

def connect_to_db(dbname, user, password, host='localhost', port=5432):
    """
    Establishes a connection to the PostgreSQL database.

    Args:
    - dbname (str): The name of the database to connect to.
    - user (str): The username used to authenticate.
    - password (str): The password used to authenticate.
    - host (str): Database host address (defaults to 'localhost').
    - port (int): Port number (defaults to 5432).

    Returns:
    - conn: A connection object to the PostgreSQL database.
    """
    connection = psycopg2.connect(dbname=dbname,
                                  user=user,
                                  password=password,
                                  host=host,
                                  port=port)
    return connection

class CustomFieldParser(FieldParser):
    """
    A custom field parser for handling special cases in DBF file data fields.
    
    This class extends the default FieldParser from dbfread to provide
    specialized handling of numeric fields ('N') that may contain
    unusual data, such as binary zeros or data that could raise a
    ValueError when being parsed in the standard way.
    """
    
    def parseN(self, field, data):
        """
        Parses numeric fields ('N') in DBF files, handling special cases.
        
        This method overrides the default numeric field parsing to address
        cases where the data might be composed of binary zeros or other
        formats that could cause the standard parsing to fail. In such cases,
        it aims to return a more sensible default (None) rather than raising an error.
        
        Args:
            field: The field descriptor.
            data: The raw bytes data for the field.
        
        Returns:
            The parsed value as a number if it can be parsed normally, or None
            if the data is composed of binary zeros or if a ValueError occurs
            during parsing.
        """
        try:
            # If the data is composed entirely of binary zeros, return None
            if data.strip(b'\x00').strip() == b'':
                return None
            return super(CustomFieldParser, self).parseN(field, data)
        except ValueError:
            # Handle any other ValueError by returning None
            return None

def clean_values(value, field_type):
    """
    Cleans values for insertion into the database.

    Args:
    - value: The value to be cleaned.
    - field_type: The type of the field.

    Returns:
    - The cleaned value.
    """
    if field_type == 'TIMESTAMP' and value == '':
        return None
    return value

def create_table_from_dbf(connection, dbf_path, table_name):
    """
    Creates or updates a PostgreSQL table based on the .dbf
    file structure and inserts the data.
    """
    dbf = DBF(dbf_path, parserclass=CustomFieldParser)
    data_types = {
        'time': 'TIMESTAMP',
        'default': 'NUMERIC'  # Default for other columns
    }

    with connection.cursor() as cursor:
        # Check if the table exists
        cursor.execute(
            "SELECT EXISTS (SELECT FROM pg_tables WHERE tablename = %s);",
            (table_name,)
        )
        table_exists = cursor.fetchone()[0]

        if not table_exists:
            # Create table if not exists
            column_defs = ', '.join(
                f"{field.name} {data_types.get(field.name.lower(), 'NUMERIC')}"
                for field in dbf.fields
            )
            cursor.execute(
                f"CREATE TABLE {table_name} ({column_defs});"
            )
        else:
            # Alter table to add new columns if necessary
            for field in dbf.fields:
                field_name = field.name
                field_type = data_types.get(field.name.lower(), 'NUMERIC')
                cursor.execute(
                    f"ALTER TABLE {table_name} ADD COLUMN IF NOT EXISTS "
                    f"{field_name} {field_type};"
                )

        connection.commit()

        # Insert data into the table
        for record in dbf:
            # Clean values
            cleaned_values = [
                clean_values(value, data_types.get(field.lower(), 'NUMERIC'))
                for field, value in record.items()
            ]

            columns = ', '.join(record.keys())
            placeholders = ', '.join(['%s'] * len(record))
            insert_query = (
                f"INSERT INTO {table_name} ({columns}) VALUES ({placeholders})"
            )
            try:
                cursor.execute(insert_query, cleaned_values)
            except UndefinedColumn as e:
                print(f"Error: {e}")
                cursor.execute(
                    f"ALTER TABLE {table_name} ADD COLUMN {e.column_name} 'NUMERIC';"
                )
                cursor.execute(insert_query, cleaned_values)

        connection.commit()

def main():
    """
    Main function to process .dbf files and insert them
    into a PostgreSQL database.
    """
    # Database connection parameters - update these
    dbname = 'YourDatabaseName'
    user = 'postgres'
    password = 'YourPassword'
    host = 'localhost'
    port = 5432

    # Path to the folder containing .dbf files
    folder_path = "Path/To/Folder/Containing/.dbf/Files"

    # The name of the table to be created/updated
    table_name = "YourTable"

    # Establish database connection
    conn = connect_to_db(dbname, user, password, host, port)

    # Process each .dbf file in the folder
    for filename in os.listdir(folder_path):
        if filename.endswith('.dbf'):
            dbf_path = os.path.join(folder_path, filename)
            create_table_from_dbf(conn, dbf_path, table_name)
            print(f"Processed {filename}")

    # Close the connection
    conn.close()

if __name__ == '__main__':
    main()