77import time
88from contextlib import contextmanager
99
10- import psycopg2
10+ import psycopg
1111from google .cloud import bigquery
12- from psycopg2 . extras import execute_values
12+ from psycopg import sql
1313
1414from pypistats .extensions import celery
1515
@@ -157,6 +157,16 @@ def process_batch_to_sqlite(cursor, table, rows):
157157 return False
158158
159159
160+ def copy_rows (cursor , table , rows , columns = ("date" , "package" , "category" , "downloads" )):
161+ """Bulk-load rows within the caller's transaction using PostgreSQL COPY."""
162+ query = sql .SQL ("COPY {} ({}) FROM STDIN" ).format (
163+ sql .Identifier (table ), sql .SQL (", " ).join (map (sql .Identifier , columns ))
164+ )
165+ with cursor .copy (query ) as copy :
166+ for row in rows :
167+ copy .write_row (row )
168+
169+
160170def transfer_sqlite_to_postgres (sqlite_cursor , date ):
161171 """Transfer all data from SQLite to PostgreSQL in a single atomic transaction."""
162172 pg_conn , pg_cursor = get_connection_cursor ()
@@ -208,12 +218,7 @@ def transfer_sqlite_to_postgres(sqlite_cursor, date):
208218 break
209219
210220 # Insert this chunk into PostgreSQL
211- insert_query = f"""
212- INSERT INTO { table } (date, package, category, downloads)
213- VALUES %s
214- """
215- # Smaller page_size to reduce memory usage when building SQL
216- execute_values (pg_cursor , insert_query , chunk , page_size = 1000 )
221+ copy_rows (pg_cursor , table , chunk )
217222
218223 chunks_transferred += 1
219224 offset += TRANSFER_CHUNK_SIZE
@@ -228,7 +233,7 @@ def transfer_sqlite_to_postgres(sqlite_cursor, date):
228233
229234 return True
230235
231- except psycopg2 .Error as e :
236+ except psycopg .Error as e :
232237 print (f"Error during PostgreSQL transfer: { e } " )
233238 pg_conn .rollback ()
234239 return False
@@ -444,15 +449,11 @@ def update_table(connection, cursor, table, rows, date):
444449 print (delete_query )
445450 cursor .execute (delete_query )
446451
447- insert_query = f"""INSERT INTO { table } (date, package, category, downloads)
448- VALUES %s"""
449-
450452 try :
451- print (insert_query )
452- execute_values (cursor , insert_query , rows )
453+ copy_rows (cursor , table , rows )
453454 connection .commit ()
454455 return True
455- except psycopg2 .IntegrityError as e :
456+ except psycopg .IntegrityError as e :
456457 connection .rollback ()
457458 return False
458459
@@ -478,18 +479,15 @@ def update_all_package_stats(date=None):
478479
479480 delete_query = f"""DELETE FROM { table }
480481 WHERE date = %s and package = '__all__'"""
481- insert_query = f"""INSERT INTO { table } (date, package, category, downloads)
482- VALUES %s"""
483482 try :
484483 print (delete_query )
485484 cursor .execute (delete_query , (date ,))
486- print (insert_query )
487485 if values : # Only insert if there are values
488- execute_values (cursor , insert_query , values )
486+ copy_rows (cursor , table , values )
489487 print (f"Inserted { len (values )} __all__ records into { table } " )
490488 connection .commit ()
491489 success [table ] = True
492- except psycopg2 .IntegrityError as e :
490+ except psycopg .IntegrityError as e :
493491 print (f"Error updating __all__ for { table } : { e } " )
494492 connection .rollback ()
495493 success [table ] = False
@@ -534,16 +532,13 @@ def update_recent_stats(date=None):
534532
535533 delete_query = f"""DELETE FROM { recent_table }
536534 WHERE category = '{ period } '"""
537- insert_query = f"""INSERT INTO { recent_table }
538- (package, category, downloads) VALUES %s"""
539535 try :
540536 print (delete_query )
541537 cursor .execute (delete_query )
542- print (insert_query )
543- execute_values (cursor , insert_query , values )
538+ copy_rows (cursor , recent_table , values , columns = ("package" , "category" , "downloads" ))
544539 connection .commit ()
545540 success [period ] = True
546- except psycopg2 .IntegrityError as e :
541+ except psycopg .IntegrityError as e :
547542 connection .rollback ()
548543 success [period ] = False
549544
@@ -554,7 +549,7 @@ def update_recent_stats(date=None):
554549
555550def get_connection_cursor ():
556551 """Get a db connection cursor."""
557- connection = psycopg2 .connect (os .environ ["DATABASE_URL" ])
552+ connection = psycopg .connect (os .environ ["DATABASE_URL" ])
558553 cursor = connection .cursor ()
559554 return connection , cursor
560555
@@ -582,7 +577,7 @@ def purge_old_data(date=None):
582577 cursor .execute (delete_query )
583578 connection .commit ()
584579 success [table ] = True
585- except psycopg2 .IntegrityError as e :
580+ except psycopg .IntegrityError as e :
586581 connection .rollback ()
587582 success [table ] = False
588583
@@ -594,7 +589,7 @@ def purge_old_data(date=None):
594589def vacuum_analyze ():
595590 """Vacuum and analyze the db."""
596591 connection , cursor = get_connection_cursor ()
597- connection .set_isolation_level ( 0 )
592+ connection .autocommit = True
598593
599594 results = {}
600595 start = time .time ()
0 commit comments