Merge pull request #30 from oxford-pharmacoepi/anto_server1

Anto server1

Author: adelmestri

8_load_source_denominator.py

@@ -31,7 +31,7 @@ def main():
 			while qa.lower() not in ['y', 'n', 'yes', 'no']:
 				qa = input('I did not understand that. Are you sure you want to DROP the ' + database_type.upper() + ' denominators tables (y/n):') 
 			if qa.lower() in ['y', 'yes']:
-				fname = dir_sql + '7a_' + database_type + '_denom_drop.sql'
+				fname = dir_sql + '8a_' + database_type + '_denom_drop.sql'
 				print('Calling ' + fname + ' ...')
 				ret = mapping_util.execute_sql_file_parallel(db_conf, fname, False)
 			if ret == True:
@@ -45,7 +45,7 @@ def main():
 # ---------------------------------------------------------
 # Create denominators tables
 # ---------------------------------------------------------
-					fname = dir_sql + '7b_' + database_type + '_denom_create.sql'
+					fname = dir_sql + '8b_' + database_type + '_denom_create.sql'
 					print('Calling ' + fname + ' ...')
 					ret = mapping_util.execute_sql_file_parallel(db_conf, fname, False)
 # ---------------------------------------------------------
@@ -72,7 +72,7 @@ def main():
 # Build PKs & IDXs
 # ---------------------------------------------------------
 					print('Build PKs and IDXs ...')
-					sql_file_list = sorted(glob.iglob(dir_sql + '7c_' + database_type + '_denom_pk_idx_*.sql'))
+					sql_file_list = sorted(glob.iglob(dir_sql + '8c_' + database_type + '_denom_pk_idx_*.sql'))
 					ret = mapping_util.execute_sql_files_parallel(db_conf, sql_file_list, True)
 					if ret == True:
 						print('Finished adding ' + database_type.upper() + ' PKs/indexes')

9_load_achilles_dqd.py

@@ -56,6 +56,8 @@ def main():
 					if ret == True:
 						data_provider = db_conf['data_provider']
 						prefix = ''
+						with_quotes = False
+						null_string = ''
 						if data_provider == 'cprd':
 							extension = '.txt'
 							separator = '	'
@@ -65,6 +67,8 @@ def main():
 						elif data_provider == 'thin':
 							extension = '.csv'
 							separator = ','
+							with_quotes = True
+							null_string = 'NA'
 						elif data_provider == 'ukbiobank':
 							extension = '.tsv'
 							separator = '	'
@@ -73,7 +77,7 @@ def main():
 						print(tbl_list_full)
 						file_list = [[dir_data + '*' + tbl + '*' + extension] for tbl in tbl_list]
 						print(file_list)
-						ret = mapping_util.load_files_parallel(db_conf, result_schema, tbl_list, file_list, dir_data_processed, separator)
+						ret = mapping_util.load_files_parallel(db_conf, result_schema, tbl_list, file_list, dir_data_processed, separator, with_quotes, null_string)
 						if ret == True:
 							print('Finished loading cdm vocabulary.')
 # ---------------------------------------------------------

mapping_util.py

@@ -158,7 +158,7 @@ def does_tbl_exist(cnx, tbl_name):
 	return(ret, exist)	
 
 # ---------------------------------------------------------
-def load_files(db_conf, schema, tbl_name, file_list, dir_processed, separator, with_quotes):
+def load_files(db_conf, schema, tbl_name, file_list, dir_processed, separator, with_quotes, null_string):
 	"Load files into tables"
 # ---------------------------------------------------------
 	ret = True
@@ -195,15 +195,17 @@ def load_files(db_conf, schema, tbl_name, file_list, dir_processed, separator, w
 				stream = StringIO()
 				if data_provider == 'ukbiobank':
 					stream.write(open(fname, encoding='cp1252', errors = 'ignore').read().replace('\\', ''))
+#				elif data_provider == 'thin':
+#					stream.write(open(fname, errors = 'ignore').read().replace('\\', '').replace(',NA,', ',,'))
 				else:
 					stream.write(open(fname, errors = 'ignore').read().replace('\\', ''))
 #				stream.write(open(fname, errors = 'ignore').read().replace('\\', '').replace('\u0000', ''))
 				stream.seek(0)
 				stream.readline()	#To avoid headers
 				if with_quotes == False:
-					cursor1.copy_from(stream, tbl_name, sep = separator, null = '')
+					cursor1.copy_from(stream, tbl_name, sep = separator, null = null_string)
 				else:
-					cursor1.copy_expert("COPY " + tbl_name + " FROM STDIN WITH (FORMAT CSV, delimiter '" + separator + "', quote '\"')", stream)
+					cursor1.copy_expert("COPY " + tbl_name + " FROM STDIN WITH (FORMAT CSV, delimiter '" + separator + "', quote '\"', NULL '" + null_string + "')", stream)
 # ---------------------------------------------------------
 # Move loaded file to PROCESSED directory
 # ---------------------------------------------------------
@@ -222,7 +224,7 @@ def load_files(db_conf, schema, tbl_name, file_list, dir_processed, separator, w
 	return(ret)
 
 # ---------------------------------------------------------
-def load_files_parallel(db_conf, schema, tbl_list, file_list, dir_processed, separator = '	', with_quotes = False):
+def load_files_parallel(db_conf, schema, tbl_list, file_list, dir_processed, separator = '	', with_quotes = False, null_string = ''):
 	"Load files into tables"
 # ---------------------------------------------------------
 	ret = True
@@ -234,7 +236,7 @@ def load_files_parallel(db_conf, schema, tbl_list, file_list, dir_processed, sep
 # Load files in parallel (all tables), sequentially within each table
 # ---------------------------------------------------------
 		with ProcessPoolExecutor(int(db_conf['max_workers'])) as executor:
-			futures = [executor.submit(load_files, db_conf, schema, tbl_name, file_list[idx], dir_processed, separator, with_quotes) for idx, tbl_name in enumerate(tbl_list)]
+			futures = [executor.submit(load_files, db_conf, schema, tbl_name, file_list[idx], dir_processed, separator, with_quotes, null_string) for idx, tbl_name in enumerate(tbl_list)]
 			for future in as_completed(futures):
 				if future.result() == False:
 					ret = False

sql_scripts/7c_aurum_denom_pk_idx_practice.sql

@@ -0,0 +1,3 @@
+drop table if exists {SOURCE_SCHEMA}.gold_acceptable_pats CASCADE;
+drop table if exists {SOURCE_SCHEMA}.gold_allpractices CASCADE;
+drop table if exists {SOURCE_SCHEMA}.gold_eligibility CASCADE;

sql_scripts/7a_aurum_denom_drop.sql sql_scripts/8a_aurum_denom_drop.sql

@@ -18,11 +18,11 @@ CREATE TABLE IF NOT EXISTS {SOURCE_SCHEMA}.aurum_acceptablepats (
 	uts date,
 	lcd date,
 	region int
-);
+) TABLESPACE pg_default;
 
 CREATE TABLE IF NOT EXISTS {SOURCE_SCHEMA}.aurum_practices (
 	pracid int,
 	lcd date,
 	uts date,
 	region int
-);
+) TABLESPACE pg_default;

sql_scripts/8a_gold_denom_drop.sql

@@ -0,0 +1,47 @@
+CREATE SCHEMA IF NOT EXISTS {SOURCE_SCHEMA};
+CREATE SCHEMA IF NOT EXISTS {SOURCE_NOK_SCHEMA};
+CREATE SCHEMA IF NOT EXISTS temp;
+CREATE SCHEMA IF NOT EXISTS results;
+
+CREATE TABLE IF NOT EXISTS {SOURCE_SCHEMA}.gold_acceptable_pats (
+	patid bigint,
+	gender char(1),
+	yob int,
+	mob int,
+	frd date,
+	crd date,
+	regstat int DEFAULT NULL,
+	reggap int DEFAULT NULL,
+	internal smallint DEFAULT NULL,
+	tod date DEFAULT NULL,
+	toreason smallint DEFAULT NULL,
+	deathdate date
+) TABLESPACE pg_default;
+
+CREATE TABLE IF NOT EXISTS {SOURCE_SCHEMA}.gold_allpractices (
+	pracid int,
+	region int,
+	lcd date,
+	uts date
+) TABLESPACE pg_default;
+
+patid	pracid	linkdate	hes_apc_e	ons_death_e	lsoa_e	sgss_e	chess_e	hes_op_e	hes_ae_e	hes_did_e	cr_e	sact_e	rtds_e	mhds_e	icnarc_e
+
+CREATE TABLE IF NOT EXISTS {SOURCE_SCHEMA}.gold_eligibility (
+	patid bigint,
+	pracid integer,
+	linkdare date,
+	hes_apc_e smallint,
+	ons_death_e smallint,
+	lsoa_e smallint,
+	sgss_e smallint,
+	chess_e smallint,
+	hes_op_e smallint,
+	hes_ae_e smallint,
+	hes_did_e smallint,
+	cr_e smallint,
+	sact_e smallint,
+	rtds_e smallint,
+	mhds_e smallint,
+	icnarc_e smallint
+) TABLESPACE pg_default;

sql_scripts/7b_aurum_denom_create.sql sql_scripts/8b_aurum_denom_create.sql

@@ -0,0 +1 @@
+alter table {SOURCE_SCHEMA}.aurum_practices add constraint pk_denom_prac primary key (pracid) USING INDEX TABLESPACE pg_default;

sql_scripts/8b_gold_denom_create.sql

@@ -0,0 +1 @@
+alter table {SOURCE_SCHEMA}.gold_acceptable_pats add constraint pk_accept_pat primary key (patid, gender, yob) USING INDEX TABLESPACE pg_default;

sql_scripts/7c_aurum_denom_pk_idx_accept_patient.sql sql_scripts/8c_aurum_denom_pk_idx_accept_patient.sql

@@ -0,0 +1 @@
+alter table {SOURCE_SCHEMA}.gold_allpractices add constraint pk_denom_prac primary key (pracid) USING INDEX TABLESPACE pg_default;