7
\$\begingroup\$

I was hoping to get some feedback on the implementation of this class for programmatically converting an access file into a PostgreSQL schema (works by obtaining schema data from the cursor object).

import os
import json
import pyodbc
import psycopg2
import time
import sys

# Refer to https://code.google.com/p/pyodbc/wiki/Cursor for information on
# cursor.tables and cursor.columns field names


class Converter:

    def __init__(self, access_con_string, pg_con_string, print_SQL):

        self.access_cur = pyodbc.connect(access_con_string).cursor()

        self.pg_con = psycopg2.connect(pg_con_string)
        self.pg_cur = self.pg_con.cursor()

        self.print_SQL = print_SQL

        self.schema_name = self.get_access_db_name()

    def get_access_db_name(self):

        # The full path of the database is stored in the table information
        # We can parse it to get the file name (to use as scheme_name)
        for table in self.access_cur.tables():
            return os.path.splitext(os.path.basename(table.table_cat))[0]

    def create_schema(self):

        SQL = """
        CREATE SCHEMA "{schema_name}"
        """.format(schema_name=self.schema_name)

        if self.print_SQL:
            print SQL

        self.pg_cur.execute(SQL)
        self.pg_con.commit()

    def create_tables(self):

        # Generate list of tables in schema
        table_list = list()
        for table in self.access_cur.tables():
            if table.table_type == "TABLE":
                table_list += [table.table_name, ]

        for table in table_list:
            SQL = """
            CREATE TABLE "{schema}"."{table}"
            (
            """.format(schema=self.schema_name, table=table)

            SQL += self.create_fields(table)

            SQL += """
            ) """

            if self.print_SQL:
                print SQL

            self.pg_cur.execute(SQL)
            self.pg_con.commit()

    def create_fields(self, table):

        postgresql_fields = {
            'COUNTER': 'serial',  # autoincrement
            'VARCHAR': 'text',  # text
            'LONGCHAR': 'text',  # memo
            'BYTE': 'integer',  # byte
            'SMALLINT': 'integer',  # integer
            'INTEGER': 'bigint',  # long integer
            'REAL': 'real',  # single
            'DOUBLE': 'double precision',  # double
            'DATETIME': 'timestamp',  # date/time
            'CURRENCY': 'money',  # currency
            'BIT':  'boolean',  # yes/no
        }

        SQL = ""
        field_list = list()
        for column in self.access_cur.columns(table=table):
            if column.type_name in postgresql_fields:
                field_list += ['"' + column.column_name + '"' +
                               " " + postgresql_fields[column.type_name], ]
            elif column.type_name == "DECIMAL":
                field_list += ['"' + column.column_name + '"' +
                               " numeric(" + str(column.column_size) + "," +
                               str(column.decimal_digits) + ")", ]
            else:
                print "column " + table + "." + column.column_name +
                " has uncatered for type: " + column.type_name

        return ",\n ".join(field_list)

    def insert_data(self):

        # Generate list of tables in schema
        table_list = list()
        for table in self.access_cur.tables():
            if table.table_type == "TABLE":
                table_list += [table.table_name, ]

        for table in table_list:
            data = self.get_access_data(table)

            # check that data exists
            if data != []:
                # Create format string (eg (%s,%s,%s)
                # the same size as the number of fields)
                format_string = "(" + ",".join(["%s", ]*len(data[0])) + ")\n"

                # pre-bind the arguments before executing - for speed
                args_string = ','.join(self.pg_cur.mogrify(format_string, x)
                                       for x in data)

                SQL = """INSERT INTO "{schema_name}"."{table_name}"
                VALUES {value_list}""".format(schema_name=self.schema_name,
                                              table_name=table,
                                              value_list=args_string)

                if self.print_SQL:
                    print SQL

                self.pg_cur.execute(SQL)

                self.pg_con.commit()

    def get_access_data(self, table):

        SQL = """SELECT *
        FROM {table_name}""".format(table_name=table)

        self.access_cur.execute(SQL)

        rows = self.access_cur.fetchall()

        data = list()
        for row in rows:
            data += [row, ]

        return data

Specifically I feel the create_fields method is a little weak in that I have an if condition to handle the DECIMAL type (as it has parameters). Would there be a better way? Style tips welcome as well!

\$\endgroup\$
5
\$\begingroup\$

First off, if you're not using Python 3.x, you need to explicitly inherit from object like this: class MyClass(object). If you are using Python 3.x, then you're fine.

If you're going to format your queries based on user input, then the following piece of code may have certain issues, like SQL injection. See this for more information.

SQL = """
CREATE TABLE "{schema}"."{table}"
(
""".format(schema=self.schema_name, table=table)

SQL += self.create_fields(table)

SQL += """
) "

The above block of code can also be shortened to the below:

SQL = "\nCREATE TABLE \"{schema}\".\"{table}\"\n(".format(schema=self.schema_name, table=table)
SQL += self.create_fields(table)
SQL += "\n) "

There are also other places in your code like the code block above the above code block that can be shortened as well.

Finally, you should also add some docstrings to your functions, rather than using inline comments, #. For example:

def my_func( ... ):
    """
    Add a description of your function and
    it's arguments here.
    """
    pass
\$\endgroup\$

Your Answer

By clicking “Post Your Answer”, you agree to our terms of service, privacy policy and cookie policy

Not the answer you're looking for? Browse other questions tagged or ask your own question.