121 lines
5.1 KiB
Python
121 lines
5.1 KiB
Python
#!/usr/bin/env python3
|
|
"""
|
|
Cleanup script to delete test vehicles from the database.
|
|
|
|
Deletes vehicles that belong to the test user (tester_pro@profibot.hu)
|
|
OR have status 'draft' OR have license plate starting with "TEST".
|
|
|
|
Handles foreign key constraints by deleting child records first.
|
|
"""
|
|
|
|
import asyncio
|
|
import sys
|
|
import os
|
|
|
|
# Add the backend directory to the path so we can import app modules
|
|
sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
|
|
|
|
from sqlalchemy import text, delete, select, and_, or_
|
|
from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession
|
|
from sqlalchemy.orm import sessionmaker
|
|
|
|
from app.core.config import settings
|
|
|
|
async def cleanup_test_vehicles():
|
|
"""Delete test vehicles and related records."""
|
|
# Create async engine
|
|
engine = create_async_engine(settings.DATABASE_URL, echo=False)
|
|
async_session = sessionmaker(engine, class_=AsyncSession, expire_on_commit=False)
|
|
|
|
deleted_counts = {
|
|
'asset_events': 0,
|
|
'asset_assignments': 0,
|
|
'asset_telemetry': 0,
|
|
'asset_financials': 0,
|
|
'asset_costs': 0,
|
|
'asset_inspections': 0,
|
|
'asset_reviews': 0,
|
|
'vehicle_expenses': 0,
|
|
'vehicle_logbook': 0,
|
|
'vehicle_ownership_history': 0,
|
|
'vehicle_transfer_requests': 0,
|
|
'service_requests': 0,
|
|
'vehicle_assets': 0
|
|
}
|
|
|
|
async with async_session() as session:
|
|
try:
|
|
# First, find all vehicle asset IDs that match our criteria
|
|
# We need to join with identity.users and identity.persons to find test user's vehicles
|
|
query = text("""
|
|
SELECT va.id
|
|
FROM vehicle.assets va
|
|
LEFT JOIN identity.persons p ON va.owner_person_id = p.id
|
|
LEFT JOIN identity.users u ON p.id = u.person_id
|
|
WHERE u.email = 'tester_pro@profibot.hu'
|
|
OR va.status = 'draft'
|
|
OR va.license_plate LIKE 'TEST%'
|
|
""")
|
|
|
|
result = await session.execute(query)
|
|
asset_ids = [row[0] for row in result.fetchall()]
|
|
|
|
if not asset_ids:
|
|
print("No test vehicles found matching criteria.")
|
|
return deleted_counts
|
|
|
|
print(f"Found {len(asset_ids)} test vehicles to delete.")
|
|
|
|
# Delete from child tables first (order matters due to foreign keys)
|
|
# List of tables with foreign keys to vehicle.assets.id and their column names
|
|
tables_to_clean = [
|
|
('fleet.asset_assignments', 'asset_id', 'asset_assignments'),
|
|
('vehicle.asset_costs', 'asset_id', 'asset_costs'),
|
|
('vehicle.asset_events', 'asset_id', 'asset_events'),
|
|
('vehicle.asset_financials', 'asset_id', 'asset_financials'),
|
|
('vehicle.asset_inspections', 'asset_id', 'asset_inspections'),
|
|
('vehicle.asset_reviews', 'asset_id', 'asset_reviews'),
|
|
('vehicle.asset_telemetry', 'asset_id', 'asset_telemetry'),
|
|
('marketplace.service_requests', 'asset_id', 'service_requests'),
|
|
('vehicle.vehicle_expenses', 'vehicle_id', 'vehicle_expenses'),
|
|
('vehicle.vehicle_logbook', 'asset_id', 'vehicle_logbook'),
|
|
('vehicle.vehicle_ownership_history', 'asset_id', 'vehicle_ownership_history'),
|
|
('vehicle.vehicle_transfer_requests', 'asset_id', 'vehicle_transfer_requests'),
|
|
]
|
|
|
|
for table, column, key in tables_to_clean:
|
|
try:
|
|
delete_query = text(f"DELETE FROM {table} WHERE {column} = ANY(:asset_ids)")
|
|
result = await session.execute(delete_query, {"asset_ids": asset_ids})
|
|
deleted_counts[key] = result.rowcount
|
|
print(f"Deleted {result.rowcount} records from {table}")
|
|
except Exception as e:
|
|
print(f"Warning: Could not delete from {table}: {e}")
|
|
# Continue with other tables
|
|
|
|
# Finally, delete from vehicle.assets
|
|
delete_assets = text("DELETE FROM vehicle.assets WHERE id = ANY(:asset_ids)")
|
|
result = await session.execute(delete_assets, {"asset_ids": asset_ids})
|
|
deleted_counts['vehicle_assets'] = result.rowcount
|
|
print(f"Deleted {result.rowcount} records from vehicle.assets")
|
|
|
|
# Commit the transaction
|
|
await session.commit()
|
|
print("Cleanup completed successfully.")
|
|
|
|
except Exception as e:
|
|
await session.rollback()
|
|
print(f"Error during cleanup: {e}")
|
|
raise
|
|
|
|
await engine.dispose()
|
|
return deleted_counts
|
|
|
|
if __name__ == "__main__":
|
|
print("Starting cleanup of test vehicles...")
|
|
counts = asyncio.run(cleanup_test_vehicles())
|
|
print("\nCleanup summary:")
|
|
for table, count in counts.items():
|
|
print(f" {table}: {count}")
|
|
total = sum(counts.values())
|
|
print(f"Total records deleted: {total}") |