62 lines
1.8 KiB
Python
62 lines
1.8 KiB
Python
import sys
|
||
import os
|
||
sys.path.insert(0, os.path.abspath(os.path.join(os.path.dirname(__file__), "../../")))
|
||
|
||
from sqlalchemy import create_engine, MetaData
|
||
from sqlalchemy.orm import sessionmaker
|
||
from config import Config
|
||
from dotenv import load_dotenv
|
||
load_dotenv()
|
||
|
||
# Źródło: SQLite
|
||
sqlite_engine = create_engine("sqlite:///instance/baza.db")
|
||
sqlite_meta = MetaData()
|
||
sqlite_meta.reflect(bind=sqlite_engine)
|
||
|
||
# Cel: PostgreSQL
|
||
pg_engine = create_engine(Config.SQLALCHEMY_DATABASE_URI)
|
||
pg_meta = MetaData()
|
||
pg_meta.reflect(bind=pg_engine)
|
||
|
||
# Sesje
|
||
SQLiteSession = sessionmaker(bind=sqlite_engine)
|
||
PGSession = sessionmaker(bind=pg_engine)
|
||
|
||
sqlite_session = SQLiteSession()
|
||
pg_session = PGSession()
|
||
|
||
def migrate_table(table_name):
|
||
print("➡️ Używana baza docelowa:", Config.SQLALCHEMY_DATABASE_URI)
|
||
print(f"\n➡️ Migruję tabelę: {table_name}")
|
||
source_table = sqlite_meta.tables.get(table_name)
|
||
target_table = pg_meta.tables.get(table_name)
|
||
|
||
if source_table is None or target_table is None:
|
||
print(f"⚠️ Pominięto: {table_name} (brak w jednej z baz)")
|
||
return
|
||
|
||
rows = sqlite_session.execute(source_table.select()).fetchall()
|
||
if not rows:
|
||
print("ℹ️ Brak danych do migracji.")
|
||
return
|
||
|
||
insert_data = [dict(row._mapping) for row in rows]
|
||
|
||
try:
|
||
with pg_engine.begin() as conn:
|
||
conn.execute(target_table.delete())
|
||
conn.execute(target_table.insert(), insert_data)
|
||
print(f"✅ Przeniesiono: {len(rows)} rekordów")
|
||
except Exception as e:
|
||
print(f"❌ Błąd przy migracji {table_name}: {e}")
|
||
|
||
|
||
def main():
|
||
tables = ["user", "shopping_list", "item", "expense", "receipt", "suggested_product"]
|
||
for table in tables:
|
||
migrate_table(table)
|
||
print("\n🎉 Migracja zakończona pomyślnie.")
|
||
|
||
if __name__ == "__main__":
|
||
main()
|