对PostgreSQL转换器的MS访问
提问于 2015-03-10 14:07:37
我希望得到一些关于该类实现的反馈，以便以编程方式将访问文件转换为PostgreSQL模式(通过从游标对象获取模式数据来工作)。
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具体来说，我觉得create_fields方法有点弱，因为我有一个if条件来处理DECIMAL类型(因为它有参数)。还有更好的办法吗？时尚小贴士也欢迎！
回答 1
Code Review用户
发布于 2015-07-04 16:39:37
首先，如果不使用Python3.x，则需要显式地从object继承如下：class MyClass(object)。如果您使用的是Python3.x，那么您就可以了。
如果要根据用户输入格式化查询，那么下面的代码可能会出现某些问题，比如SQL注入。关于更多信息，请看这个。
SQL = """
CREATE TABLE "{schema}"."{table}"
(
""".format(schema=self.schema_name, table=table)
SQL += self.create_fields(table)
SQL += """
) "上述代码块也可以缩短为以下代码：
SQL = "\nCREATE TABLE \"{schema}\".\"{table}\"\n(".format(schema=self.schema_name, table=table)
SQL += self.create_fields(table)
SQL += "\n) "您的代码中还有其他地方，如上面代码块上面的代码块，它们也可以被缩短。
最后，还应该向函数中添加一些文档字符串，而不是使用内联注释#。例如：
def my_func( ... ):
"""
Add a description of your function and
it's arguments here.
"""
pass页面原文内容由Code Review提供。腾讯云小微IT领域专用引擎提供翻译支持
原文链接：
https://codereview.stackexchange.com/questions/83800
复制相关文章
腾讯云开发者
