first pass on generating new metadata from local
This commit is contained in:
John Krauss
parent
827104756e
commit
2a1598d491
@ -1,40 +1,67 @@
|
||||
import os
|
||||
import psycopg2
|
||||
import subprocess
|
||||
|
||||
DB_CONN = psycopg2.connect('postgres://{user}:{password}@{host}:{port}/{database}'.format(
|
||||
user=os.environ.get('PGUSER', 'postgres'),
|
||||
password=os.environ.get('PGPASSWORD', ''),
|
||||
host=os.environ.get('PGHOST', 'localhost'),
|
||||
port=os.environ.get('PGPORT', '5432'),
|
||||
database=os.environ.get('PGDATABASE', 'postgres'),
|
||||
))
|
||||
CURSOR = DB_CONN.cursor()
|
||||
|
||||
|
||||
def query(q):
|
||||
'''
|
||||
Query the database.
|
||||
'''
|
||||
try:
|
||||
CURSOR.execute(q)
|
||||
return CURSOR
|
||||
except:
|
||||
DB_CONN.rollback()
|
||||
raise
|
||||
|
||||
|
||||
def commit():
|
||||
try:
|
||||
DB_CONN.commit()
|
||||
except:
|
||||
DB_CONN.rollback()
|
||||
raise
|
||||
|
||||
from sqldumpr import Dumpr
|
||||
|
||||
def get_tablename_query(column_id, boundary_id, timespan):
|
||||
"""
|
||||
given a column_id, boundary-id (us.census.tiger.block_group), and
|
||||
timespan, give back the current table hash from the data observatory
|
||||
"""
|
||||
q = """
|
||||
return """
|
||||
SELECT t.tablename, geoid_ct.colname colname
|
||||
FROM obs_table t,
|
||||
obs_column_table geoid_ct,
|
||||
obs_column_table data_ct
|
||||
FROM observatory.obs_table t,
|
||||
observatory.obs_column_table geoid_ct,
|
||||
observatory.obs_column_table data_ct
|
||||
WHERE
|
||||
t.id = geoid_ct.table_id AND
|
||||
t.id = data_ct.table_id AND
|
||||
geoid_ct.column_id =
|
||||
(SELECT source_id
|
||||
FROM obs_column_to_column
|
||||
FROM observatory.obs_column_to_column
|
||||
WHERE target_id = '{boundary_id}'
|
||||
AND reltype = 'geom_ref'
|
||||
) AND
|
||||
data_ct.column_id = '{column_id}' AND
|
||||
timespan = '{timespan}'
|
||||
""".replace('\n','')
|
||||
""".format(column_id=column_id,
|
||||
boundary_id=boundary_id,
|
||||
timespan=timespan)
|
||||
|
||||
return q.format(column_id=column_id,
|
||||
boundary_id=boundary_id,
|
||||
timespan=timespan)
|
||||
|
||||
def select_star(tablename):
|
||||
return "SELECT * FROM {}".format(tablename)
|
||||
|
||||
cdb = Dumpr('observatory.cartodb.com','')
|
||||
|
||||
metadata = ['obs_table', 'obs_column_table', 'obs_column', 'obs_column_tag',
|
||||
'obs_tag', 'obs_column_to_column', 'obs_dump_version', ]
|
||||
METADATA_TABLES = ['obs_table', 'obs_column_table', 'obs_column', 'obs_column_tag',
|
||||
'obs_tag', 'obs_column_to_column', 'obs_dump_version', 'obs_meta',
|
||||
'obs_meta_numer', 'obs_meta_denom', 'obs_meta_geom',
|
||||
'obs_meta_timespan', ]
|
||||
|
||||
fixtures = [
|
||||
('us.census.tiger.census_tract', 'us.census.tiger.census_tract', '2014'),
|
||||
@ -52,162 +79,131 @@ fixtures = [
|
||||
('us.census.tiger.block_group_clipped', 'us.census.tiger.block_group_clipped', '2014'),
|
||||
]
|
||||
|
||||
unique_tables = set()
|
||||
OUTFILE_PATH = 'src/pg/test/fixtures/load_fixtures.sql'
|
||||
DROPFILE_PATH = 'src/pg/test/fixtures/drop_fixtures.sql'
|
||||
|
||||
for f in fixtures:
|
||||
column_id, boundary_id, timespan = f
|
||||
tablename_query = get_tablename_query(*f)
|
||||
resp = cdb.query(tablename_query).json()['rows'][0]
|
||||
tablename = resp['tablename']
|
||||
colname = resp['colname']
|
||||
table_colname = (tablename, colname, boundary_id, )
|
||||
if table_colname not in unique_tables:
|
||||
print table_colname
|
||||
unique_tables.add(table_colname)
|
||||
def dump(cols, tablename, where=''):
|
||||
|
||||
print unique_tables
|
||||
with open(DROPFILE_PATH, 'a') as dropfile:
|
||||
dropfile.write('DROP TABLE IF EXISTS observatory.{tablename};\n'.format(
|
||||
tablename=tablename,
|
||||
))
|
||||
|
||||
with open('src/pg/test/fixtures/load_fixtures.sql', 'w') as outfile:
|
||||
with open('src/pg/test/fixtures/drop_fixtures.sql', 'w') as dropfiles:
|
||||
outfile.write('SET client_min_messages TO WARNING;\n\set ECHO none\n')
|
||||
dropfiles.write('SET client_min_messages TO WARNING;\n\set ECHO none\n')
|
||||
for tablename in metadata:
|
||||
cdb.dump(select_star(tablename), tablename, outfile, schema='observatory')
|
||||
dropfiles.write('DROP TABLE IF EXISTS observatory.{};\n'.format(tablename))
|
||||
print tablename
|
||||
subprocess.check_call('pg_dump -x --section=pre-data -t observatory.{tablename} '
|
||||
' | sed "s:, pg_catalog::" '
|
||||
' | sed "s:CREATE TABLE :CREATE TABLE observatory.:" '
|
||||
' | sed "s:ALTER TABLE.*OWNER.*::" '
|
||||
' >> {outfile}'.format(
|
||||
tablename=tablename,
|
||||
outfile=OUTFILE_PATH,
|
||||
), shell=True)
|
||||
|
||||
for tablename, colname, boundary_id in unique_tables:
|
||||
if 'zcta5' in boundary_id:
|
||||
where = '\'11%\''
|
||||
compare = 'LIKE'
|
||||
elif 'whosonfirst' in boundary_id:
|
||||
where = '(\'85632785\',\'85633051\',\'85633111\',\'85633147\',\'85633253\',\'85633267\')'
|
||||
compare = 'IN'
|
||||
else:
|
||||
where = '\'36047%\''
|
||||
compare = 'LIKE'
|
||||
print ' '.join([select_star(tablename), "WHERE {}::text {} {}".format(colname, compare, where)])
|
||||
cdb.dump(' '.join([select_star(tablename), "WHERE {}::text {} {}".format(colname, compare, where)]),
|
||||
tablename, outfile, schema='observatory')
|
||||
dropfiles.write('DROP TABLE IF EXISTS observatory.{};\n'.format(tablename))
|
||||
with open(OUTFILE_PATH, 'a') as outfile:
|
||||
outfile.write('COPY observatory."{}" FROM stdin WITH CSV HEADER;\n'.format(tablename))
|
||||
|
||||
subprocess.check_call('''
|
||||
psql -c "COPY (SELECT {cols} \
|
||||
FROM observatory.{tablename} {where}) \
|
||||
TO STDOUT WITH CSV HEADER" >> {outfile}'''.format(
|
||||
cols=cols,
|
||||
tablename=tablename,
|
||||
where=where,
|
||||
outfile=OUTFILE_PATH,
|
||||
), shell=True)
|
||||
|
||||
with open(OUTFILE_PATH, 'a') as outfile:
|
||||
outfile.write('\\.\n\n')
|
||||
|
||||
|
||||
outfile.write('''
|
||||
ALTER TABLE observatory.obs_table
|
||||
ADD PRIMARY KEY (id);
|
||||
ALTER TABLE observatory.obs_column_table
|
||||
ADD PRIMARY KEY (column_id, table_id);
|
||||
CREATE UNIQUE INDEX ON observatory.obs_column_table (table_id, column_id);
|
||||
CREATE UNIQUE INDEX ON observatory.obs_column_table (table_id, colname);
|
||||
ALTER TABLE observatory.obs_column
|
||||
ADD PRIMARY KEY (id);
|
||||
ALTER TABLE observatory.obs_column_to_column
|
||||
ADD PRIMARY KEY (source_id, target_id, reltype);
|
||||
CREATE UNIQUE INDEX ON observatory.obs_column_to_column (target_id, source_id, reltype);
|
||||
CREATE INDEX ON observatory.obs_column_to_column (reltype);
|
||||
ALTER TABLE observatory.obs_column_tag
|
||||
ADD PRIMARY KEY (column_id, tag_id);
|
||||
CREATE UNIQUE INDEX ON observatory.obs_column_tag (tag_id, column_id);
|
||||
ALTER TABLE observatory.obs_tag
|
||||
ADD PRIMARY KEY (id);
|
||||
CREATE INDEX ON observatory.obs_tag (type);
|
||||
def main():
|
||||
unique_tables = set()
|
||||
|
||||
VACUUM ANALYZE observatory.obs_table;
|
||||
VACUUM ANALYZE observatory.obs_column_table;
|
||||
VACUUM ANALYZE observatory.obs_column;
|
||||
VACUUM ANALYZE observatory.obs_column_to_column;
|
||||
VACUUM ANALYZE observatory.obs_column_tag;
|
||||
VACUUM ANALYZE observatory.obs_tag;
|
||||
for f in fixtures:
|
||||
column_id, boundary_id, timespan = f
|
||||
tablename_query = get_tablename_query(column_id, boundary_id, timespan)
|
||||
tablename, colname = query(tablename_query).fetchone()
|
||||
table_colname = (tablename, colname, boundary_id, )
|
||||
if table_colname not in unique_tables:
|
||||
print(table_colname)
|
||||
unique_tables.add(table_colname)
|
||||
|
||||
CREATE TABLE observatory.obs_meta AS
|
||||
SELECT numer_c.id numer_id,
|
||||
denom_c.id denom_id,
|
||||
geom_c.id geom_id,
|
||||
MAX(numer_c.name) numer_name,
|
||||
MAX(denom_c.name) denom_name,
|
||||
MAX(geom_c.name) geom_name,
|
||||
MAX(numer_c.description) numer_description,
|
||||
MAX(denom_c.description) denom_description,
|
||||
MAX(geom_c.description) geom_description,
|
||||
MAX(numer_c.aggregate) numer_aggregate,
|
||||
MAX(denom_c.aggregate) denom_aggregate,
|
||||
MAX(geom_c.aggregate) geom_aggregate,
|
||||
MAX(numer_c.type) numer_type,
|
||||
MAX(denom_c.type) denom_type,
|
||||
MAX(geom_c.type) geom_type,
|
||||
MAX(numer_data_ct.colname) numer_colname,
|
||||
MAX(denom_data_ct.colname) denom_colname,
|
||||
MAX(geom_geom_ct.colname) geom_colname,
|
||||
MAX(numer_geomref_ct.colname) numer_geomref_colname,
|
||||
MAX(denom_geomref_ct.colname) denom_geomref_colname,
|
||||
MAX(geom_geomref_ct.colname) geom_geomref_colname,
|
||||
MAX(numer_t.tablename) numer_tablename,
|
||||
MAX(denom_t.tablename) denom_tablename,
|
||||
MAX(geom_t.tablename) geom_tablename,
|
||||
MAX(numer_t.timespan) numer_timespan,
|
||||
MAX(denom_t.timespan) denom_timespan,
|
||||
MAX(numer_c.weight) numer_weight,
|
||||
MAX(denom_c.weight) denom_weight,
|
||||
MAX(geom_c.weight) geom_weight,
|
||||
MAX(geom_t.timespan) geom_timespan,
|
||||
MAX(geom_t.the_geom_webmercator)::geometry AS the_geom_webmercator,
|
||||
ARRAY_AGG(DISTINCT s_tag.id) section_tags,
|
||||
ARRAY_AGG(DISTINCT ss_tag.id) subsection_tags,
|
||||
ARRAY_AGG(DISTINCT unit_tag.id) unit_tags
|
||||
FROM observatory.obs_column_table numer_data_ct,
|
||||
observatory.obs_table numer_t,
|
||||
observatory.obs_column_table numer_geomref_ct,
|
||||
observatory.obs_column geomref_c,
|
||||
observatory.obs_column_to_column geomref_c2c,
|
||||
observatory.obs_column geom_c,
|
||||
observatory.obs_column_table geom_geom_ct,
|
||||
observatory.obs_column_table geom_geomref_ct,
|
||||
observatory.obs_table geom_t,
|
||||
observatory.obs_column_tag ss_ctag,
|
||||
observatory.obs_tag ss_tag,
|
||||
observatory.obs_column_tag s_ctag,
|
||||
observatory.obs_tag s_tag,
|
||||
observatory.obs_column_tag unit_ctag,
|
||||
observatory.obs_tag unit_tag,
|
||||
observatory.obs_column numer_c
|
||||
LEFT JOIN (
|
||||
observatory.obs_column_to_column denom_c2c
|
||||
JOIN observatory.obs_column denom_c ON denom_c2c.target_id = denom_c.id
|
||||
JOIN observatory.obs_column_table denom_data_ct ON denom_data_ct.column_id = denom_c.id
|
||||
JOIN observatory.obs_table denom_t ON denom_data_ct.table_id = denom_t.id
|
||||
JOIN observatory.obs_column_table denom_geomref_ct ON denom_geomref_ct.table_id = denom_t.id
|
||||
) ON denom_c2c.source_id = numer_c.id
|
||||
WHERE numer_c.id = numer_data_ct.column_id
|
||||
AND numer_data_ct.table_id = numer_t.id
|
||||
AND numer_t.id = numer_geomref_ct.table_id
|
||||
AND numer_geomref_ct.column_id = geomref_c.id
|
||||
AND geomref_c2c.reltype = 'geom_ref'
|
||||
AND geomref_c.id = geomref_c2c.source_id
|
||||
AND geom_c.id = geomref_c2c.target_id
|
||||
AND geom_geomref_ct.column_id = geomref_c.id
|
||||
AND geom_geomref_ct.table_id = geom_t.id
|
||||
AND geom_geom_ct.column_id = geom_c.id
|
||||
AND geom_geom_ct.table_id = geom_t.id
|
||||
AND geom_c.type ILIKE 'geometry'
|
||||
AND numer_c.type NOT ILIKE 'geometry'
|
||||
AND numer_t.id != geom_t.id
|
||||
AND numer_c.id != geomref_c.id
|
||||
AND unit_tag.type = 'unit'
|
||||
AND ss_tag.type = 'subsection'
|
||||
AND s_tag.type = 'section'
|
||||
AND unit_ctag.column_id = numer_c.id
|
||||
AND unit_ctag.tag_id = unit_tag.id
|
||||
AND ss_ctag.column_id = numer_c.id
|
||||
AND ss_ctag.tag_id = ss_tag.id
|
||||
AND s_ctag.column_id = numer_c.id
|
||||
AND s_ctag.tag_id = s_tag.id
|
||||
AND (denom_c2c.reltype = 'denominator' OR denom_c2c.reltype IS NULL)
|
||||
AND (denom_geomref_ct.column_id = geomref_c.id OR denom_geomref_ct.column_id IS NULL)
|
||||
AND (denom_t.timespan = numer_t.timespan OR denom_t.timespan IS NULL)
|
||||
GROUP BY numer_c.id, denom_c.id, geom_c.id,
|
||||
numer_t.id, denom_t.id, geom_t.id;
|
||||
''')
|
||||
print unique_tables
|
||||
|
||||
dropfiles.write('''
|
||||
DROP TABLE IF EXISTS observatory.obs_meta;
|
||||
''')
|
||||
with open(OUTFILE_PATH, 'w') as outfile:
|
||||
outfile.write('SET client_min_messages TO WARNING;\n\\set ECHO none\n')
|
||||
outfile.write('CREATE SCHEMA IF NOT EXISTS observatory;\n\n')
|
||||
|
||||
with open(DROPFILE_PATH, 'w') as dropfile:
|
||||
dropfile.write('SET client_min_messages TO WARNING;\n\\set ECHO none\n')
|
||||
|
||||
for tablename in METADATA_TABLES:
|
||||
print(tablename)
|
||||
if tablename == 'obs_meta':
|
||||
where = "WHERE " + " OR ".join([
|
||||
"(numer_id, geom_id, numer_timespan) = ('{}', '{}', '{}')".format(
|
||||
numer_id, geom_id, timespan)
|
||||
for numer_id, geom_id, timespan in fixtures
|
||||
])
|
||||
elif tablename == 'obs_meta_numer':
|
||||
where = "WHERE " + " OR ".join([
|
||||
"numer_id IN ('{}', '{}')".format(numer_id, geom_id)
|
||||
for numer_id, geom_id, timespan in fixtures
|
||||
])
|
||||
elif tablename == 'obs_meta_denom':
|
||||
where = "WHERE " + " OR ".join([
|
||||
"denom_id IN ('{}', '{}')".format(numer_id, geom_id)
|
||||
for numer_id, geom_id, timespan in fixtures
|
||||
])
|
||||
elif tablename == 'obs_meta_geom':
|
||||
where = "WHERE " + " OR ".join([
|
||||
"geom_id IN ('{}', '{}')".format(numer_id, geom_id)
|
||||
for numer_id, geom_id, timespan in fixtures
|
||||
])
|
||||
elif tablename == 'obs_meta_timespan':
|
||||
where = "WHERE " + " OR ".join([
|
||||
"timespan_id = ('{}')".format(timespan)
|
||||
for numer_id, geom_id, timespan in fixtures
|
||||
])
|
||||
elif tablename == 'obs_column':
|
||||
where = "WHERE " + " OR ".join([
|
||||
"id IN ('{}', '{}')".format(numer_id, geom_id)
|
||||
for numer_id, geom_id, timespan in fixtures
|
||||
])
|
||||
elif tablename in ('obs_column_table', 'obs_column_tag'):
|
||||
where = "WHERE " + " OR ".join([
|
||||
"column_id IN ('{}', '{}')".format(numer_id, geom_id)
|
||||
for numer_id, geom_id, timespan in fixtures
|
||||
])
|
||||
elif tablename == 'obs_column_to_column':
|
||||
where = "WHERE " + " OR ".join([
|
||||
"source_id IN ('{}', '{}') OR target_id IN ('{}', '{}')".format(
|
||||
numer_id, geom_id, numer_id, geom_id)
|
||||
for numer_id, geom_id, timespan in fixtures
|
||||
])
|
||||
elif tablename == 'obs_table':
|
||||
where = "WHERE " + " OR ".join([
|
||||
"timespan = '{}'".format(timespan)
|
||||
for numer_id, geom_id, timespan in fixtures
|
||||
])
|
||||
else:
|
||||
where = ''
|
||||
dump('*', tablename, where)
|
||||
|
||||
for tablename, colname, boundary_id in unique_tables:
|
||||
if 'zcta5' in boundary_id:
|
||||
where = '\'11%\''
|
||||
compare = 'LIKE'
|
||||
elif 'whosonfirst' in boundary_id:
|
||||
where = '(\'85632785\',\'85633051\',\'85633111\',\'85633147\',\'85633253\',\'85633267\')'
|
||||
compare = 'IN'
|
||||
else:
|
||||
where = '\'36047%\''
|
||||
compare = 'LIKE'
|
||||
print ' '.join(['*', tablename, "WHERE {}::text {} {}".format(colname, compare, where)])
|
||||
dump('*', tablename, "WHERE {}::text {} {}".format(colname, compare, where))
|
||||
#cdb.dump(' '.join([select_star(tablename), "WHERE {}::text {} {}".format(colname, compare, where)]),
|
||||
# tablename, outfile, schema='observatory')
|
||||
#dropfiles.write('DROP TABLE IF EXISTS observatory.{};\n'.format(tablename))
|
||||
|
||||
if __name__ == '__main__':
|
||||
main()
|
||||
|
||||
7
src/pg/test/fixtures/drop_fixtures.sql
vendored
7
src/pg/test/fixtures/drop_fixtures.sql
vendored
@ -7,6 +7,11 @@ DROP TABLE IF EXISTS observatory.obs_column_tag;
|
||||
DROP TABLE IF EXISTS observatory.obs_tag;
|
||||
DROP TABLE IF EXISTS observatory.obs_column_to_column;
|
||||
DROP TABLE IF EXISTS observatory.obs_dump_version;
|
||||
DROP TABLE IF EXISTS observatory.obs_meta;
|
||||
DROP TABLE IF EXISTS observatory.obs_meta_numer;
|
||||
DROP TABLE IF EXISTS observatory.obs_meta_denom;
|
||||
DROP TABLE IF EXISTS observatory.obs_meta_geom;
|
||||
DROP TABLE IF EXISTS observatory.obs_meta_timespan;
|
||||
DROP TABLE IF EXISTS observatory.obs_65f29658e096ca1485bf683f65fdbc9f05ec3c5d;
|
||||
DROP TABLE IF EXISTS observatory.obs_1746e37b7cd28cb131971ea4187d42d71f09c5f3;
|
||||
DROP TABLE IF EXISTS observatory.obs_1a098da56badf5f32e336002b0a81708c40d29cd;
|
||||
@ -20,5 +25,3 @@ DROP TABLE IF EXISTS observatory.obs_6c1309a64d8f3e6986061f4d1ca7b57743e75e74;
|
||||
DROP TABLE IF EXISTS observatory.obs_d39f7fe5959891c8296490d83c22ded31c54af13;
|
||||
DROP TABLE IF EXISTS observatory.obs_144e8b4f906885b2e057ac4842644a553ae49c6e;
|
||||
DROP TABLE IF EXISTS observatory.obs_c6fb99c47d61289fbb8e561ff7773799d3fcc308;
|
||||
|
||||
DROP TABLE IF EXISTS observatory.obs_meta;
|
||||
|
||||
37722
src/pg/test/fixtures/load_fixtures.sql
vendored
37722
src/pg/test/fixtures/load_fixtures.sql
vendored
File diff suppressed because one or more lines are too long
Loading…
Reference in New Issue
Block a user