Catalog Plugin

pip install 'daita-agents[postgresql]'    # PostgreSQL discovery
pip install 'daita-agents[mysql]'         # MySQL discovery
pip install 'daita-agents[mongodb]'       # MongoDB discovery
pip install 'daita-agents[aws]'           # AWS infrastructure discovery
pip install 'daita-agents[azure]'         # Azure infrastructure discovery
pip install 'daita-agents[opensearch]'    # OpenSearch discovery
pip install 'daita-agents[github]'        # GitHub repository discovery

from daita import Agent
from daita.plugins import catalog
 
cat = catalog()
 
agent = Agent(
    name="Schema Analyst",
    prompt="You are a database schema expert. Help users discover and analyze database structures.",
    tools=[cat]
)
 
await agent.start()
result = await agent.run("Discover the PostgreSQL database schema at localhost")

from daita.plugins import catalog
 
cat = catalog()
 
result = await cat.discover_postgres(
    connection_string="postgresql://user:pass@localhost:5432/mydb",
    schema="public"
)
schema = result['schema']
 
print(f"Found {schema['table_count']} tables")
print(f"Total columns: {schema['column_count']}")

catalog(
    backend: Optional[Any] = None,
    organization_id: Optional[int] = None,
    auto_persist: bool = False
)

result = await cat.discover_postgres(
    connection_string="postgresql://user:pass@localhost:5432/analytics",
    schema="public",
    ssl_mode="verify-full",  # or "require" for pgbouncer poolers
    persist=False
)
schema = result['schema']
 
print(f"Database: {schema['database_type']}")
print(f"Tables: {schema['table_count']}")
for table in schema['tables']:
    print(f"  - {table['table_name']}: {table['row_count']} rows")

result = await cat.discover_mysql(
    connection_string="mysql://user:pass@localhost:3306/sales",
    schema="sales"
)

result = await cat.discover_mongodb(
    connection_string="mongodb://localhost:27017",
    database="analytics",
    sample_size=100
)

result = await cat.discover_openapi(
    spec_url="https://api.example.com/openapi.json",
    service_name="Example API"
)

from daita.plugins.catalog import CatalogPlugin
from daita.plugins.catalog.aws import AWSDiscoverer
from daita.plugins.catalog.azure import AzureDiscoverer
from daita.plugins.catalog.gcp import GCPDiscoverer
from daita.plugins.catalog.github import GitHubScanner
 
cat = CatalogPlugin(auto_persist=True)
 
# Register discoverers
cat.add_discoverer(AWSDiscoverer(regions=["us-east-1", "us-west-2"]))
cat.add_discoverer(AzureDiscoverer(
    subscriptions=["00000000-0000-0000-0000-000000000000"],
    locations=["eastus", "westus2"],
))
cat.add_discoverer(GCPDiscoverer(
    projects=["my-project-prod", "my-project-staging"],
    credentials_path="service-account.json",     # or use ADC
))
cat.add_discoverer(GitHubScanner(org="my-company", token="ghp_..."))
 
# Discover all infrastructure
result = await cat.discover_all(concurrency=5)
 
print(f"Found {len(result.stores)} stores")
for store in result.stores:
    print(f"  {store.display_name} ({store.store_type}) — {store.region}")
 
if result.errors:
    for err in result.errors:
        print(f"  Error in {err.discoverer_name}: {err.error}")

from daita.plugins.catalog.azure import AzureDiscoverer
 
cat.add_discoverer(AzureDiscoverer(
    subscriptions=["00000000-0000-0000-0000-000000000000"],
    locations=["eastus"],
    services=["sql", "postgresql", "cosmosdb", "blob", "eventhub"],
))

from daita.plugins.catalog.profiler import build_default_profilers
 
# Register profilers for all supported store types
for profiler in build_default_profilers():
    cat.add_profiler(profiler)
 
# Profile a specific store
store = result.stores[0]
profiler = cat._find_profiler(store.store_type)
schema = await profiler.profile(store)
 
print(f"Tables: {len(schema.tables)}")
for table in schema.tables:
    print(f"  {table.name}: {len(table.columns)} columns")

result = await cat.discover_and_profile(concurrency=5)

from daita.plugins.catalog import BaseDiscoverer, DiscoveredStore
 
class MyDiscoverer(BaseDiscoverer):
    name = "my-source"
 
    async def authenticate(self):
        # Set up credentials
        ...
 
    async def enumerate(self):
        # Yield DiscoveredStore instances
        yield DiscoveredStore(
            id="my-store-1",
            store_type="postgresql",
            display_name="My Database",
            source="my-source",
            region="us-east-1",
        )
 
cat.add_discoverer(MyDiscoverer())

from daita.plugins.catalog import BaseProfiler, NormalizedSchema
 
class MyProfiler(BaseProfiler):
    def supports(self, store_type: str) -> bool:
        return store_type == "my-store-type"
 
    async def profile(self, store) -> NormalizedSchema:
        # Connect and extract schema
        ...
 
cat.add_profiler(MyProfiler())

result_dev = await cat.discover_postgres(connection_string="postgresql://localhost/dev_db")
result_prod = await cat.discover_postgres(connection_string="postgresql://localhost/prod_db")
 
result = await cat.compare_schemas(result_dev['schema'], result_prod['schema'])
comparison = result['comparison']
 
print(f"Added tables: {comparison['added_tables']}")
print(f"Removed tables: {comparison['removed_tables']}")
print(f"Modified columns: {len(comparison['modified_columns'])}")
print(f"Breaking changes: {comparison['breaking_changes']}")

# Via agent (recommended)
result = await agent.run("Compare store abc123 against its baseline")

result = await cat.discover_postgres(connection_string="postgresql://localhost/mydb")
 
# Export as Mermaid ER diagram
mermaid = await cat.export_diagram(result['schema'], format="mermaid")
print(mermaid['diagram'])
 
# Also supports "json_schema" format

from daita import Agent
from daita.plugins import catalog
 
cat = catalog(auto_persist=True)
 
agent = Agent(
    name="Schema Expert",
    prompt="You are a database schema analyst. Help users discover and understand database structures.",
    tools=[cat]
)
 
await agent.start()
 
result = await agent.run("""
Discover the PostgreSQL schema at localhost:5432/analytics.
Then compare it to the MySQL schema at localhost:3306/sales.
Identify any structural differences.
""")
 
await agent.stop()

from daita.plugins.catalog import register_catalog_backend_factory
 
register_catalog_backend_factory(lambda: MyStorageBackend())

result = await cat.prune_stale_catalog(max_age_seconds=604800)  # 7 days
print(f"Removed stale entries: {result['removed']}")