cBUCZSYw

1import pandas as pd
2import yaml
3import sqlalchemy
4import migrate
5from sqlalchemy import select, cast, column, create_engine, Table, MetaData, insert
6from sqlalchemy.schema import DDLElement, DDL
7from sqlalchemy.ext.compiler import compiles
8from sqlalchemy.sql.expression import ColumnClause
9
10sqlalchemy.__version__
11
12# load in driver
13with open("property_driver_2.yml") as f:
14    driver_2 = yaml.load(f)
15    
16   
17engine = create_engine("postgresql://educkworth:Skyline18!@outra1.cfvp8mszcntj.eu-west-1.rds.amazonaws.com/outra")
18metadata = MetaData(bind=engine, reflect=True, schema='educkworth')
19
20table_X = metadata.tables["educkworth.tab_property_100"]
21
22
23class AlterColumn(DDLElement):
24
25    def __init__(self, column, cmd):
26        self.column = column
27        self.cmd = cmd
28        
29
30@compiles(AlterColumn, 'postgresql')
31def alter_enum_dtype(element, compiler, cats, **kw):
32    enum_name = element.name + "_enum_%d"
33    i = 0
34    while True:
35        enum_name_i = enum_name % i
36        result = engine.execute("select exists (select 1 from pg_type where typname = '%s');" % enum_name_i).fetchone()[0]
37        if result == False:
38            break
39        i += 1    
40    enum_name = enum_name_i
41    return 'CREATE TYPE %s as enum(%s); \
42            ALTER TABLE %s ALTER COLUMN %s TYPE %s USING %s::%s;' % (
43        enum_name,
44        cats,
45        element.table.name,
46        element.name,
47        enum_name,
48        element.name,
49        enum_name)
50    
51@compiles(AlterColumn, 'postgresql')
52def alter_col_dtype(element, compiler, dtype, **kw):
53        return "ALTER TABLE %s ALTER COLUMN %s TYPE %s USING %s::%s;" % (
54            element.table.name,
55            element.name,
56            dtype,
57            element.name,
58            dtype)
59    
60@compiles(AlterColumn, 'postgresql')
61def alter_int_dtype(element, complier, **kw):
62    return "ALTER TABLE %s ALTER COLUMN %s TYPE BIGINT USING %s::numeric::BIGINT;" % (
63            element.table.name,
64            element.name,
65            element.name)
66
67@compiles(AlterColumn, 'postgresql')
68def alter_str_dtype(element, **kw):
69    return "ALTER TABLE %s ALTER COLUMN %s TYPE TEXT; "  % (
70            element.table.name,
71            element.name)
72
73@compiles(AlterColumn, 'postgresql')
74def map_string(element, before, after, **kw):
75    return "UPDATE %s SET %s = replace(%s, '%s', '%s'); " % (
76            element.table.name, element.name, element.name, before, after)
77  
78def datatype_update(col, table, schema_dictionary, type_dictionary):
79    tab_c = table.columns[col]
80    c_dtype = schema_dictionary[col]['dtype']
81    sql_dtype = type_dictionary[c_dtype]
82    postgres_dtype = table.columns[col].type
83    
84    map_sql = ""
85    sql = ""
86    
87    # Map, must convert to string first
88    # update postgres type to text
89    if 'map' in schema_dictionary[col]:
90        if str(postgres_dtype).find("enum") > -1:
91            map_sql += alter_str_dtype(table.columns[col])
92            postgres_dtype = "TEXT"
93        for key in schema_dictionary[col]['map']:
94            map_sql += map_string(table.columns[col], key, schema_dictionary[col]['map'][key])
95
96    # If already in the correct format, no need to udpate        
97    if str(sql_dtype) == str(postgres_dtype):
98        sql = ""
99    
100    # Create Enums for categories
101    elif c_dtype == 'cat':    
102        if str(postgres_dtype).find("enum") == -1:            
103            c_cats = "'"+"', '".join(schema_dictionary[col]['cat'])+"'"
104            sql = alter_enum_dtype(table.columns[col], engine, cats=c_cats)
105        else:
106            sql = ""
107    
108    # For integers need to convert to numeric then integers
109    elif c_dtype == 'int':
110        sql = alter_int_dtype(table.columns[col], engine)
111    #     
112    else:
113        sql = alter_col_dtype(table.columns[col], engine, dtype=sql_dtype)
114            
115    sql = map_sql + sql
116            
117    return sql
118
119type_dict = {'int':'BIGINT',
120            'text': 'TEXT',
121            'bool': 'boolean',
122            'cat':'Enum',
123            'float': 'real',
124            'date': 'DATE'}
125            
126for col in table_X.columns:
127    print(col.name)
128    if col.name in driver_2.keys():
129        sql= datatype_update(col.name, table_X, driver_2, type_dict)
130        if sql == "":
131            print("column update not required")
132        else:
133            print(sql)
134            result = engine.execute(sql) 
135    else:
136        print(col.name, " is not in driver. Please update driver.")