leocrm-bot
4c7f09c518
- I001: import block sorting in main.py, routes/__init__.py, companies.py, conftest.py, check_indexes.py - F401: unused imports removed (asyncio, UTC, datetime, AsyncSession) - 6 ASYNC240/230 warnings remain (os.path.exists in async, open() in test) — minor, non-blocking
89 lines
2.5 KiB
Python
89 lines
2.5 KiB
Python
#!/usr/bin/env python3
|
|
"""Check database indexes — verify all required indexes exist for performance.
|
|
|
|
Usage:
|
|
python scripts/check_indexes.py
|
|
|
|
Checks that critical indexes for contacts, companies, and tenant scoping
|
|
are present in the database.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import asyncio
|
|
import os
|
|
import sys
|
|
|
|
# Add project root to path
|
|
sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
|
|
|
|
from sqlalchemy import text
|
|
from sqlalchemy.ext.asyncio import create_async_engine
|
|
|
|
from app.config import get_settings
|
|
|
|
# Required indexes for performance (table, index_name)
|
|
REQUIRED_INDEXES = [
|
|
("contacts", "ix_contacts_tenant_deleted"),
|
|
("contacts", "ix_contacts_tenant_name"),
|
|
("contacts", "ix_contacts_email"),
|
|
("companies", "ix_companies_tenant_deleted"),
|
|
("companies", "ix_companies_tenant_name"),
|
|
("companies", "ix_companies_industry"),
|
|
("company_contacts", "ix_cc_company"),
|
|
("company_contacts", "ix_cc_contact"),
|
|
]
|
|
|
|
|
|
async def check_indexes() -> int:
|
|
"""Check that all required indexes exist in the database.
|
|
|
|
Returns 0 if all indexes present, 1 if any missing.
|
|
"""
|
|
settings = get_settings()
|
|
engine = create_async_engine(settings.database_url)
|
|
|
|
missing: list[tuple[str, str]] = []
|
|
present: list[tuple[str, str]] = []
|
|
|
|
async with engine.connect() as conn:
|
|
for table_name, index_name in REQUIRED_INDEXES:
|
|
result = await conn.execute(
|
|
text(
|
|
"SELECT 1 FROM pg_indexes "
|
|
"WHERE schemaname = 'public' "
|
|
"AND tablename = :table "
|
|
"AND indexname = :index"
|
|
),
|
|
{"table": table_name, "index": index_name},
|
|
)
|
|
if result.scalar():
|
|
present.append((table_name, index_name))
|
|
print(f" ✓ {table_name}.{index_name}")
|
|
else:
|
|
missing.append((table_name, index_name))
|
|
print(f" ✗ MISSING: {table_name}.{index_name}")
|
|
|
|
await engine.dispose()
|
|
|
|
print(f"\nSummary: {len(present)} present, {len(missing)} missing")
|
|
if missing:
|
|
print("\nMissing indexes:")
|
|
for table, idx in missing:
|
|
print(f" - {table}.{idx}")
|
|
return 1
|
|
|
|
print("\nAll required indexes present!")
|
|
return 0
|
|
|
|
|
|
def main():
|
|
"""Run index check."""
|
|
print("Checking database indexes...")
|
|
exit_code = asyncio.run(check_indexes())
|
|
sys.exit(exit_code)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|