70 lines
2.8 KiB
Python
Executable File
70 lines
2.8 KiB
Python
Executable File
import asyncio
|
|
from sqlalchemy import inspect, text
|
|
from sqlalchemy.ext.asyncio import create_async_engine
|
|
|
|
# CSERÉLD KI a saját adataidra, ha nem ezeket használod!
|
|
DATABASE_URL = "postgresql+asyncpg://kincses:MiskociA74@postgres-db:5432/service_finder"
|
|
|
|
async def run_inspection():
|
|
# Aszinkron motor létrehozása
|
|
engine = create_async_engine(DATABASE_URL)
|
|
|
|
async with engine.connect() as conn:
|
|
def do_inspect(sync_conn):
|
|
inspector = inspect(sync_conn)
|
|
|
|
# 1. Sémák lekérése (amiket mi hoztunk létre + public)
|
|
target_schemas = ['public', 'data', 'ref']
|
|
all_schemas = inspector.get_schema_names()
|
|
schemas_to_check = [s for s in target_schemas if s in all_schemas]
|
|
|
|
print(f"{'='*60}")
|
|
print(f" JÁRMŰNYILVÁNTARTÓ RENDSZER - ADATBÁZIS AUDIT")
|
|
print(f"{'='*60}")
|
|
|
|
for schema in schemas_to_check:
|
|
print(f"\n--- SÉMA: {schema.upper()} ---")
|
|
tables = inspector.get_table_names(schema=schema)
|
|
|
|
if not tables:
|
|
print(" (Nincsenek táblák ebben a sémában)")
|
|
continue
|
|
|
|
for table_name in tables:
|
|
print(f"\n[Tábla: {schema}.{table_name}]")
|
|
|
|
# Oszlopok részletei
|
|
columns = inspector.get_columns(table_name, schema=schema)
|
|
for col in columns:
|
|
pk = " [PK]" if col['primary_key'] else ""
|
|
nullable = "NULL" if col['nullable'] else "NOT NULL"
|
|
print(f" L {col['name']:<15} | {str(col['type']):<12} | {nullable}{pk}")
|
|
|
|
# Idegen kulcsok (Kapcsolatok)
|
|
fks = inspector.get_foreign_keys(table_name, schema=schema)
|
|
for fk in fks:
|
|
constrained = fk['constrained_columns']
|
|
referred_schema = fk['referred_schema']
|
|
referred_table = fk['referred_table']
|
|
referred_cols = fk['referred_columns']
|
|
print(f" --> Kapcsolat: {constrained} -> {referred_schema}.{referred_table}({referred_cols})")
|
|
|
|
# Mivel az SQLAlchemy inspector alapvetően szinkron, run_sync-et használunk
|
|
await conn.run_sync(do_inspect)
|
|
|
|
# 2. Extra: Kiterjesztések ellenőrzése
|
|
print(f"\n{'='*60}")
|
|
print(" AKTÍV POSTGRESQL KITERJESZTÉSEK:")
|
|
result = await conn.execute(text("SELECT extname FROM pg_extension;"))
|
|
for row in result:
|
|
print(f" - {row[0]}")
|
|
print(f"{'='*60}")
|
|
|
|
await engine.dispose()
|
|
|
|
if __name__ == "__main__":
|
|
try:
|
|
asyncio.run(run_inspection())
|
|
except Exception as e:
|
|
print(f"HIBA TÖRTÉNT: {e}")
|