feat: add db apis

.gitignore

@@ -24,6 +24,8 @@ share/python-wheels/
 *.egg
 MANIFEST
 
+*collection.json
+
 *.manifest
 *.spec
 
@@ -117,3 +119,5 @@ logs/
 
 *.tmp
 *.temp
+
+local_deploy.py

Dockerfile

@@ -1,7 +1,7 @@
 FROM python:3.12-slim
 
-LABEL maintainer="reconciliation-non-ai-extractor"
-LABEL description="Enterprise document extraction API with Microsoft MarkItDown, RapidOCR, and spaCy"
+LABEL maintainer="All API Collection"
+LABEL description="All API Collection - Document extraction, conversion, and database query API"
 LABEL version="1.0.0"
 
 RUN apt-get update && apt-get install -y --no-install-recommends \

README.md

@@ -1,5 +1,5 @@
 ---
-title: MarkItDown API — v2.1.0
+title: All API Collection — v2.1.0
 emoji: ⚡
 colorFrom: red
 colorTo: purple

__init__.py

@@ -1,4 +0,0 @@
-from __future__ import annotations
-
-__version__ = "1.0.0"
-__app_name__ = "reconciliation-non-ai-extractor"

app/api/deps.py

@@ -5,6 +5,7 @@ from fastapi import Depends
 from app.core.security import require_api_key
 from app.services.auth_service import AuthService
 from app.services.converter_service import ConverterService
+from app.services.database_service import DatabaseService
 from app.services.extraction_service import ExtractionService
 from app.services.ocr_service import OCRService
 
@@ -25,5 +26,9 @@ def get_extraction_service() -> ExtractionService:
     return ExtractionService()
 
 
+def get_database_service() -> DatabaseService:
+    return DatabaseService()
+
+
 def require_auth(token: str = Depends(require_api_key)) -> str:
     return token

app/api/server.py

@@ -8,6 +8,7 @@ from fastapi.middleware.cors import CORSMiddleware
 from fastapi.middleware.gzip import GZipMiddleware
 
 from app.config import get_settings
+from app.core.database import pool_manager
 from app.core.logger import get_logger
 from app.api.v1.router import api_v1_router
 
@@ -35,12 +36,14 @@ async def _self_ping():
 async def lifespan(app: FastAPI):
     asyncio.create_task(_self_ping())
     yield
+    _logger.info("Shutting down database connection pools...")
+    await pool_manager.close_all()
 
 
 def create_application() -> FastAPI:
     app = FastAPI(
         title=_settings.app_name,
-        description="Document-to-Markdown and structured data extraction API powered by Microsoft MarkItDown, RapidOCR, and spaCy.",
+        description="All API Collection - Document extraction, conversion, and database query API.",
         version=_settings.app_version,
         docs_url="/docs",
         redoc_url="/redoc",
@@ -72,7 +75,7 @@ def create_application() -> FastAPI:
 
     @app.get("/ping", include_in_schema=False)
     async def ping():
-        return {"message": f"{_settings.app_name} is running..."}
+        return {"name": f"{_settings.app_name}", "version": _settings.app_version}
 
     return app
 

app/api/v1/database.py

@@ -0,0 +1,48 @@
+from __future__ import annotations
+
+from typing import Annotated
+
+from fastapi import APIRouter, Depends, HTTPException, status
+
+from app.api.deps import require_auth
+from app.core.logger import get_logger
+from app.models.schemas import DatabaseQueryRequest, DatabaseQueryResponse
+from app.services.database_service import DatabaseService
+
+router = APIRouter()
+_logger = get_logger(__name__)
+
+
+@router.post(
+    "/database/query",
+    response_model=DatabaseQueryResponse,
+    summary="Execute queries against MySQL, PostgreSQL, or MongoDB",
+)
+async def execute_database_query(
+    body: DatabaseQueryRequest,
+    token: Annotated[str, Depends(require_auth)],
+    db_service: Annotated[DatabaseService, Depends()] = None,
+) -> DatabaseQueryResponse:
+    if db_service is None:
+        db_service = DatabaseService()
+
+    _logger.info(
+        "Database query request: type=%s, %s",
+        body.db_type,
+        body.connection.safe_repr(),
+    )
+
+    try:
+        return await db_service.execute_query(body)
+    except HTTPException:
+        raise
+    except Exception as exc:
+        _logger.error("Unexpected error processing database query: %s", exc)
+        raise HTTPException(
+            status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
+            detail={
+                "success": False,
+                "execution_time_ms": 0,
+                "error": {"message": f"Internal error: {exc}", "code": "INTERNAL_ERROR"},
+            },
+        )

app/api/v1/router.py

@@ -2,9 +2,10 @@ from __future__ import annotations
 
 from fastapi import APIRouter
 
-from app.api.v1 import batch, convert, system
+from app.api.v1 import batch, convert, database, system
 
 api_v1_router = APIRouter()
 api_v1_router.include_router(convert.router, tags=["Convert"])
 api_v1_router.include_router(batch.router, tags=["Batch"])
 api_v1_router.include_router(system.router, tags=["System"])
+api_v1_router.include_router(database.router, tags=["Database"])

app/config.py

@@ -13,7 +13,7 @@ class Settings(BaseSettings):
         extra="ignore",
     )
 
-    app_name: str = "MarkItDown API"
+    app_name: str = "All API Collection"
     app_version: str = "1.0.0"
     environment: str = "production"
     host: str = "0.0.0.0"

app/core/database/__init__.py

@@ -0,0 +1,15 @@
+from app.core.database.base import BaseExecutor, ConnectionConfig, StatementResult
+from app.core.database.mysql import MySQLExecutor
+from app.core.database.postgresql import PostgreSQLExecutor
+from app.core.database.mongodb import MongoDBExecutor
+from app.core.database.pool import pool_manager
+
+__all__ = [
+    "BaseExecutor",
+    "ConnectionConfig",
+    "StatementResult",
+    "MySQLExecutor",
+    "PostgreSQLExecutor",
+    "MongoDBExecutor",
+    "pool_manager",
+]

app/core/database/base.py

@@ -0,0 +1,137 @@
+from __future__ import annotations
+
+import asyncio
+import time
+from abc import ABC, abstractmethod
+from dataclasses import dataclass, field
+from typing import Any
+
+from app.core.logger import get_logger
+
+_logger = get_logger(__name__)
+
+
+@dataclass(frozen=True)
+class ConnectionConfig:
+    db_type: str
+    host: str
+    port: int
+    database: str
+    username: str
+    password: str
+    ssl_enabled: bool = False
+    ssl_ca_cert: str | None = None
+    ssl_cert: str | None = None
+    ssl_key: str | None = None
+    query_timeout_seconds: float = 30.0
+    connection_timeout_seconds: float = 10.0
+    max_rows: int = 10000
+
+    @property
+    def pool_key(self) -> str:
+        return f"{self.db_type}:{self.username}@{self.host}:{self.port}/{self.database}"
+
+    @property
+    def safe_repr(self) -> str:
+        return (
+            f"ConnectionConfig(db_type={self.db_type}, host={self.host}, "
+            f"port={self.port}, database={self.database}, "
+            f"username={self.username}, ssl={self.ssl_enabled})"
+        )
+
+
+@dataclass
+class StatementResult:
+    success: bool
+    rows: int = 0
+    data: list[dict[str, Any]] = field(default_factory=list)
+    error: str | None = None
+    error_code: str | None = None
+
+
+class BaseExecutor(ABC):
+    def __init__(self, config: ConnectionConfig) -> None:
+        self._config = config
+        self._pool: Any = None
+        self._closed = False
+        self._lock = asyncio.Lock()
+        self._max_connection_retries = 3
+
+    @abstractmethod
+    async def _create_pool(self) -> Any:
+        ...
+
+    @abstractmethod
+    async def _execute_queries(
+        self, pool: Any, queries: list[Any], use_transaction: bool
+    ) -> list[StatementResult]:
+        ...
+
+    async def execute(
+        self, queries: list[Any], use_transaction: bool = True
+    ) -> list[StatementResult]:
+        if self._closed:
+            raise RuntimeError("Executor has been closed")
+        pool = await self._get_or_create_pool()
+        return await self._execute_queries(pool, queries, use_transaction)
+
+    async def _get_or_create_pool(self) -> Any:
+        async with self._lock:
+            if self._pool is not None:
+                return self._pool
+
+            last_exc: Exception | None = None
+            for attempt in range(self._max_connection_retries):
+                try:
+                    self._pool = await asyncio.wait_for(
+                        self._create_pool(),
+                        timeout=self._config.connection_timeout_seconds,
+                    )
+                    _logger.info(
+                        "Created pool for %s (attempt %d)",
+                        self._config.safe_repr, attempt + 1,
+                    )
+                    return self._pool
+                except asyncio.TimeoutError:
+                    last_exc = TimeoutError(
+                        f"Connection timed out after {self._config.connection_timeout_seconds}s"
+                    )
+                    _logger.warning(
+                        "Pool creation timeout for %s (attempt %d)",
+                        self._config.safe_repr, attempt + 1,
+                    )
+                except Exception as exc:
+                    last_exc = exc
+                    _logger.warning(
+                        "Pool creation failed for %s (attempt %d): %s",
+                        self._config.safe_repr, attempt + 1, exc,
+                    )
+
+                if attempt < self._max_connection_retries - 1:
+                    wait = 0.1 * (2**attempt)
+                    await asyncio.sleep(wait)
+
+            msg = (
+                f"Failed to create connection pool after {self._max_connection_retries} attempts"
+            )
+            if last_exc is not None:
+                msg += f": {last_exc}"
+            raise RuntimeError(msg) from last_exc
+
+    async def close(self) -> None:
+        async with self._lock:
+            if self._closed:
+                return
+            self._closed = True
+            if self._pool is not None:
+                await self._close_pool(self._pool)
+                self._pool = None
+                _logger.info("Closed pool for %s", self._config.safe_repr)
+
+    @abstractmethod
+    async def _close_pool(self, pool: Any) -> None:
+        ...
+
+    @property
+    def is_closed(self) -> bool:
+        return self._closed

app/core/database/mongodb.py

@@ -0,0 +1,105 @@
+from __future__ import annotations
+
+from typing import Any
+from urllib.parse import quote_plus
+
+from motor.motor_asyncio import AsyncIOMotorClient, AsyncIOMotorDatabase
+
+from app.core.database.base import BaseExecutor, ConnectionConfig, StatementResult
+from app.core.logger import get_logger
+
+_logger = get_logger(__name__)
+
+_ATLAS_SRV_DOMAINS = frozenset({"mongodb.net", "mongodbatlas.com"})
+_DATABASE_LEVEL_MARKER = "$database"
+
+
+class MongoDBExecutor(BaseExecutor):
+    async def _create_pool(self) -> AsyncIOMotorClient:
+        client_kwargs: dict[str, Any] = {
+            "serverSelectionTimeoutMS": int(
+                self._config.connection_timeout_seconds * 1000
+            ),
+            "connectTimeoutMS": int(self._config.connection_timeout_seconds * 1000),
+            "maxPoolSize": 10,
+            "minPoolSize": 1,
+        }
+
+        if self._config.ssl_enabled:
+            client_kwargs["tls"] = True
+
+        host = self._config.host
+
+        if self._is_atlas_srv(host):
+            return self._connect_via_uri(client_kwargs)
+
+        client_kwargs["host"] = host
+        client_kwargs["port"] = self._config.port
+        client_kwargs["username"] = self._config.username
+        client_kwargs["password"] = self._config.password
+
+        return AsyncIOMotorClient(**client_kwargs)
+
+    def _connect_via_uri(self, client_kwargs: dict[str, Any]) -> AsyncIOMotorClient:
+        escaped_user = quote_plus(self._config.username)
+        escaped_pass = quote_plus(self._config.password)
+        uri = (
+            f"mongodb+srv://{escaped_user}:{escaped_pass}@{self._config.host}/"
+            f"{self._config.database}?retryWrites=true&w=majority"
+        )
+        client_kwargs.pop("tls", None)
+        return AsyncIOMotorClient(uri, **client_kwargs)
+
+    @staticmethod
+    def _is_atlas_srv(host: str) -> bool:
+        host_lower = host.lower()
+        for domain in _ATLAS_SRV_DOMAINS:
+            if domain in host_lower:
+                return True
+        return host_lower.startswith("mongodb+srv://")
+
+    async def _execute_queries(
+        self,
+        pool: AsyncIOMotorClient,
+        queries: list[dict[str, Any]],
+        use_transaction: bool,
+    ) -> list[StatementResult]:
+        db: AsyncIOMotorDatabase = pool[self._config.database]
+        results: list[StatementResult] = []
+        for query_item in queries:
+            collection_name = query_item.get("collection", _DATABASE_LEVEL_MARKER)
+            pipeline = query_item.get("pipeline", [])
+            try:
+                if collection_name == _DATABASE_LEVEL_MARKER:
+                    cursor = db.aggregate(pipeline)
+                else:
+                    cursor = db[collection_name].aggregate(pipeline)
+
+                data = []
+                async for doc in cursor:
+                    if "_id" in doc:
+                        doc["_id"] = str(doc["_id"])
+                    data.append(doc)
+                    if len(data) >= self._config.max_rows:
+                        break
+                results.append(
+                    StatementResult(success=True, rows=len(data), data=data)
+                )
+            except Exception as exc:
+                results.append(
+                    StatementResult(
+                        success=False,
+                        error=str(exc),
+                        error_code=type(exc).__name__,
+                    )
+                )
+                break
+        return results
+
+    async def _close_pool(self, pool: AsyncIOMotorClient) -> None:
+        pool.close()
+
+    async def _get_or_create_pool(self) -> AsyncIOMotorClient:
+        if self._pool is not None:
+            return self._pool
+        return await super()._get_or_create_pool()

app/core/database/mysql.py

@@ -0,0 +1,101 @@
+from __future__ import annotations
+
+import asyncio
+import sys
+from typing import Any
+
+import aiomysql
+
+from app.core.database.base import BaseExecutor, ConnectionConfig, StatementResult
+from app.core.logger import get_logger
+
+_logger = get_logger(__name__)
+
+
+class MySQLExecutor(BaseExecutor):
+    async def _create_pool(self) -> aiomysql.Pool:
+        return await aiomysql.create_pool(
+            host=self._config.host,
+            port=self._config.port,
+            user=self._config.username,
+            password=self._config.password,
+            db=self._config.database,
+            minsize=1,
+            maxsize=10,
+            autocommit=False,
+            connect_timeout=self._config.connection_timeout_seconds,
+            pool_recycle=3600,
+        )
+
+    async def _get_or_create_pool(self) -> aiomysql.Pool:
+        async with self._lock:
+            if self._pool is not None:
+                return self._pool
+
+            last_error: Exception | None = None
+            for attempt in range(self._max_connection_retries):
+                try:
+                    self._pool = await asyncio.wait_for(
+                        self._create_pool(),
+                        timeout=self._config.connection_timeout_seconds,
+                    )
+                    return self._pool
+                except Exception as exc:
+                    last_error = exc
+                    _logger.warning(
+                        "MySQL connection failed (attempt %d): %s",
+                        attempt + 1, exc,
+                    )
+                    if attempt < self._max_connection_retries - 1:
+                        await asyncio.sleep(0.1 * (2**attempt))
+
+            raise RuntimeError(
+                f"Failed to connect to MySQL after {self._max_connection_retries} attempts"
+            ) from last_error
+
+    async def _execute_queries(
+        self, pool: aiomysql.Pool, queries: list[str], use_transaction: bool
+    ) -> list[StatementResult]:
+        async with pool.acquire() as conn:
+            if use_transaction:
+                await conn.begin()
+            try:
+                results: list[StatementResult] = []
+                for query in queries:
+                    result = await self._execute_one(conn, query)
+                    results.append(result)
+                    if not result.success and use_transaction:
+                        await conn.rollback()
+                        return results
+                if use_transaction:
+                    await conn.commit()
+                return results
+            except Exception:
+                if use_transaction:
+                    try:
+                        await conn.rollback()
+                    except Exception:
+                        pass
+                raise
+
+    async def _execute_one(self, conn: Any, query: str) -> StatementResult:
+        try:
+            async with conn.cursor(aiomysql.DictCursor) as cursor:
+                await cursor.execute(query)
+                if cursor.description:
+                    rows = await cursor.fetchall()
+                    data = [dict(row) for row in rows][: self._config.max_rows]
+                    return StatementResult(success=True, rows=len(data), data=data)
+                await conn.commit()
+                return StatementResult(success=True, rows=cursor.rowcount, data=[])
+        except Exception as exc:
+            error_code = getattr(exc, "args", [None])[0]
+            if isinstance(error_code, int):
+                error_code = str(error_code)
+            else:
+                error_code = type(exc).__name__
+            return StatementResult(success=False, error=str(exc), error_code=error_code)
+
+    async def _close_pool(self, pool: aiomysql.Pool) -> None:
+        pool.close()
+        await pool.wait_closed()

app/core/database/pool.py

@@ -0,0 +1,57 @@
+from __future__ import annotations
+
+import asyncio
+from typing import Any
+
+from app.core.database.base import BaseExecutor, ConnectionConfig
+from app.core.database.mysql import MySQLExecutor
+from app.core.database.postgresql import PostgreSQLExecutor
+from app.core.database.mongodb import MongoDBExecutor
+from app.core.logger import get_logger
+
+_logger = get_logger(__name__)
+
+_EXECUTOR_MAP: dict[str, type[BaseExecutor]] = {
+    "mysql": MySQLExecutor,
+    "postgresql": PostgreSQLExecutor,
+    "mongodb": MongoDBExecutor,
+}
+
+
+class PoolManager:
+    def __init__(self) -> None:
+        self._executors: dict[str, BaseExecutor] = {}
+        self._lock = asyncio.Lock()
+        self._closed = False
+
+    async def get_executor(self, config: ConnectionConfig) -> BaseExecutor:
+        if self._closed:
+            raise RuntimeError("PoolManager has been shut down")
+
+        key = config.pool_key
+        async with self._lock:
+            if key not in self._executors:
+                executor_cls = _EXECUTOR_MAP.get(config.db_type)
+                if executor_cls is None:
+                    raise ValueError(f"Unsupported database type: {config.db_type}")
+                self._executors[key] = executor_cls(config)
+                _logger.info(
+                    "Created executor for %s (key=%s)", config.safe_repr, key,
+                )
+            return self._executors[key]
+
+    async def close_all(self) -> None:
+        async with self._lock:
+            if self._closed:
+                return
+            self._closed = True
+            for key, executor in self._executors.items():
+                try:
+                    await executor.close()
+                except Exception as exc:
+                    _logger.error("Error closing executor %s: %s", key, exc)
+            self._executors.clear()
+            _logger.info("All database executors closed")
+
+
+pool_manager = PoolManager()

app/core/database/postgresql.py

@@ -0,0 +1,93 @@
+from __future__ import annotations
+
+import ssl as ssl_module
+from typing import Any
+
+import asyncpg
+
+from app.core.database.base import BaseExecutor, ConnectionConfig, StatementResult
+from app.core.logger import get_logger
+
+_logger = get_logger(__name__)
+
+
+class PostgreSQLExecutor(BaseExecutor):
+    async def _create_pool(self) -> asyncpg.Pool:
+        ssl_ctx: ssl_module.SSLContext | None = None
+        if self._config.ssl_enabled:
+            ssl_ctx = self._build_ssl_context()
+
+        return await asyncpg.create_pool(
+            host=self._config.host,
+            port=self._config.port,
+            user=self._config.username,
+            password=self._config.password,
+            database=self._config.database,
+            min_size=1,
+            max_size=10,
+            timeout=self._config.connection_timeout_seconds,
+            ssl=ssl_ctx,
+            max_inactive_connection_lifetime=3600.0,
+        )
+
+    async def _execute_queries(
+        self, pool: asyncpg.Pool, queries: list[str], use_transaction: bool
+    ) -> list[StatementResult]:
+        async with pool.acquire() as conn:
+            if use_transaction:
+                tr = conn.transaction()
+                await tr.start()
+            try:
+                results: list[StatementResult] = []
+                for query in queries:
+                    result = await self._execute_one(conn, query)
+                    results.append(result)
+                    if not result.success and use_transaction:
+                        await tr.rollback()
+                        _logger.warning(
+                            "Query failed, transaction rolled back: %s",
+                            result.error,
+                        )
+                        return results
+                if use_transaction:
+                    await tr.commit()
+                return results
+            except Exception as exc:
+                if use_transaction:
+                    try:
+                        await tr.rollback()
+                    except Exception:
+                        pass
+                raise
+
+    async def _execute_one(self, conn: asyncpg.Connection, query: str) -> StatementResult:
+        try:
+            stripped = query.strip().upper()
+            if stripped.startswith("SELECT") or stripped.startswith("WITH"):
+                rows = await conn.fetch(query)
+                data = [dict(row) for row in rows][: self._config.max_rows]
+                return StatementResult(success=True, rows=len(data), data=data)
+
+            if "RETURNING" in stripped:
+                rows = await conn.fetch(query)
+                data = [dict(row) for row in rows][: self._config.max_rows]
+                return StatementResult(success=True, rows=len(data), data=data)
+
+            result = await conn.execute(query)
+            parts = result.split()
+            rowcount = int(parts[-1]) if parts[-1].isdigit() else 0
+            return StatementResult(success=True, rows=rowcount, data=[])
+        except Exception as exc:
+            return StatementResult(
+                success=False,
+                error=str(exc),
+                error_code=type(exc).__name__,
+            )
+
+    async def _close_pool(self, pool: asyncpg.Pool) -> None:
+        await pool.close()
+
+    @staticmethod
+    def _build_ssl_context() -> ssl_module.SSLContext:
+        ctx = ssl_module.create_default_context()
+        return ctx

app/models/schemas.py

@@ -1,8 +1,8 @@
 from __future__ import annotations
 
-from typing import Any, Dict, List, Optional
+from typing import Any, Dict, List, Literal, Optional
 
-from pydantic import BaseModel, Field, field_validator
+from pydantic import BaseModel, Field, field_validator, model_validator
 
 
 class ConversionMetadata(BaseModel):
@@ -107,3 +107,118 @@ class SpacyLabelsResponse(BaseModel):
     spacy_labels: Dict[str, str]
     source_types: Dict[str, str]
     example_mappings: Dict[str, Dict[str, Any]]
+
+
+class SSLConfig(BaseModel):
+    enabled: bool = False
+    ca_cert: Optional[str] = None
+    cert: Optional[str] = None
+    key: Optional[str] = None
+
+
+class DatabaseConnection(BaseModel):
+    host: str
+    port: Optional[int] = None
+    database: str
+    username: str
+    password: str = Field(repr=False)
+    ssl: SSLConfig = SSLConfig()
+
+    @field_validator("host")
+    @classmethod
+    def host_not_empty(cls, v: str) -> str:
+        stripped = v.strip()
+        if not stripped:
+            raise ValueError("host must not be empty")
+        return stripped
+
+    @field_validator("port")
+    @classmethod
+    def validate_port(cls, v: Optional[int]) -> Optional[int]:
+        if v is not None and (v < 1 or v > 65535):
+            raise ValueError("port must be between 1 and 65535")
+        return v
+
+    @field_validator("database")
+    @classmethod
+    def database_not_empty(cls, v: str) -> str:
+        stripped = v.strip()
+        if not stripped:
+            raise ValueError("database must not be empty")
+        return stripped
+
+    def safe_repr(self) -> str:
+        return (
+            f"host={self.host}, port={self.port}, "
+            f"database={self.database}, username={self.username}"
+        )
+
+
+class DatabaseQueryRequest(BaseModel):
+    db_type: Literal["mysql", "postgresql", "mongodb"]
+    connection: DatabaseConnection
+    query: List[Any]
+    use_transaction: bool = True
+    query_timeout_seconds: float = Field(default=10.0, ge=1.0, le=300.0)
+    connection_timeout_seconds: float = Field(default=10.0, ge=1.0, le=60.0)
+    max_rows: int = Field(default=10000, ge=1, le=1_000_000)
+
+    @model_validator(mode="after")
+    def validate_query(self) -> "DatabaseQueryRequest":
+        if not self.query:
+            raise ValueError("query must not be empty")
+        if self.db_type in ("mysql", "postgresql"):
+            for i, q in enumerate(self.query):
+                if not isinstance(q, str):
+                    raise ValueError(f"query[{i}] must be a string for {self.db_type}")
+                if not q.strip():
+                    raise ValueError(f"query[{i}] must not be empty")
+        elif self.db_type == "mongodb":
+            for i, q in enumerate(self.query):
+                if not isinstance(q, dict):
+                    raise ValueError(f"query[{i}] must be an object with 'collection' and 'pipeline'")
+                if "collection" not in q or "pipeline" not in q:
+                    raise ValueError(f"query[{i}] must have 'collection' and 'pipeline' fields")
+                if not isinstance(q["pipeline"], list):
+                    raise ValueError(f"query[{i}].pipeline must be an array")
+        return self
+
+    def to_connection_config(self) -> Dict[str, Any]:
+        return {
+            "db_type": self.db_type,
+            "host": self.connection.host,
+            "port": self.connection.port or self._default_port(),
+            "database": self.connection.database,
+            "username": self.connection.username,
+            "password": self.connection.password,
+            "ssl_enabled": self.connection.ssl.enabled,
+            "ssl_ca_cert": self.connection.ssl.ca_cert,
+            "ssl_cert": self.connection.ssl.cert,
+            "ssl_key": self.connection.ssl.key,
+            "query_timeout_seconds": self.query_timeout_seconds,
+            "connection_timeout_seconds": self.connection_timeout_seconds,
+            "max_rows": self.max_rows,
+        }
+
+    def _default_port(self) -> int:
+        return {"mysql": 3306, "postgresql": 5432, "mongodb": 27017}[self.db_type]
+
+
+class StatementResultSchema(BaseModel):
+    success: bool
+    rows: int = 0
+    data: List[Dict[str, Any]] = []
+    error: Optional[str] = None
+    error_code: Optional[str] = None
+
+
+class DatabaseQueryError(BaseModel):
+    message: str
+    code: Optional[str] = None
+
+
+class DatabaseQueryResponse(BaseModel):
+    success: bool
+    execution_time_ms: float
+    results: Optional[List[StatementResultSchema]] = None
+    error: Optional[DatabaseQueryError] = None

app/services/database_service.py

@@ -0,0 +1,103 @@
+from __future__ import annotations
+
+import time
+from typing import Any
+
+from app.core.database import ConnectionConfig, pool_manager
+from app.core.database.base import StatementResult
+from app.core.logger import get_logger
+from app.models.schemas import (
+    DatabaseQueryError,
+    DatabaseQueryRequest,
+    DatabaseQueryResponse,
+    StatementResultSchema,
+)
+
+_logger = get_logger(__name__)
+
+
+def _root_cause(exc: Exception) -> str:
+    cause = exc.__cause__ or exc.__context__
+    if cause:
+        return f"{exc} [{cause}]"
+    return str(exc)
+
+
+class DatabaseService:
+
+    async def execute_query(self, request: DatabaseQueryRequest) -> DatabaseQueryResponse:
+        start_time = time.monotonic()
+        config = ConnectionConfig(**request.to_connection_config())
+
+        try:
+            executor = await pool_manager.get_executor(config)
+        except Exception as exc:
+            elapsed = (time.monotonic() - start_time) * 1000
+            _logger.error(
+                "Failed to acquire executor for %s: %s",
+                config.safe_repr, exc,
+            )
+            return DatabaseQueryResponse(
+                success=False,
+                execution_time_ms=round(elapsed, 2),
+                error=DatabaseQueryError(
+                    message=f"Connection failed: {_root_cause(exc)}",
+                    code=type(exc).__name__,
+                ),
+            )
+
+        try:
+            results = await executor.execute(
+                request.query,
+                use_transaction=request.use_transaction,
+            )
+        except Exception as exc:
+            elapsed = (time.monotonic() - start_time) * 1000
+            _logger.error(
+                "Query execution failed for %s: %s",
+                config.safe_repr, exc,
+            )
+            return DatabaseQueryResponse(
+                success=False,
+                execution_time_ms=round(elapsed, 2),
+                error=DatabaseQueryError(
+                    message=f"Execution failed: {_root_cause(exc)}",
+                    code=type(exc).__name__,
+                ),
+            )
+
+        elapsed = (time.monotonic() - start_time) * 1000
+
+        statement_results = [
+            StatementResultSchema(
+                success=r.success,
+                rows=r.rows,
+                data=r.data,
+                error=r.error,
+                error_code=r.error_code,
+            )
+            for r in results
+        ]
+
+        overall_success = all(r.success for r in results)
+
+        if overall_success:
+            _logger.info(
+                "Query success for %s (%d stmts, %.2fms)",
+                config.safe_repr, len(results), elapsed,
+            )
+            return DatabaseQueryResponse(
+                success=True,
+                execution_time_ms=round(elapsed, 2),
+                results=statement_results,
+            )
+
+        _logger.warning(
+            "Query partial/full failure for %s (%d stmts, %.2fms)",
+            config.safe_repr, len(results), elapsed,
+        )
+        return DatabaseQueryResponse(
+            success=False,
+            execution_time_ms=round(elapsed, 2),
+            results=statement_results,
+        )

main.py

@@ -7,7 +7,7 @@ from app.core.logger import get_logger
 logger = get_logger(__name__)
 
 if __name__ == "__main__":
-    logger.info("Starting %s server", "reconciliation-non-ai-extractor")
+    logger.info("Starting %s server", "All API Collection")
     uvicorn.run(
         "app.api.server:app",
         host="0.0.0.0",

postman_collection.json

@@ -1,387 +0,0 @@
-{
-    "info": {
-        "_postman_id": "reconciliation-not-ai-extractor-service",
-        "name": "Reconciliation-not-ai-extractor-service",
-        "description": "Production-ready Postman collection for the Document-to-Markdown and structured data extraction API powered by Microsoft MarkItDown, RapidOCR, and spaCy.\n\n**Base URL:** `{{base_url}}`\n**Authentication:** API key via `x-api-key` header\n\n## Endpoints\n- **System:** Health, info, supported formats, spaCy labels\n- **Convert:** Single file and single URL to Markdown\n- **Batch:** Bulk conversion of files and URLs",
-        "schema": "https://schema.getpostman.com/json/collection/v2.1.0/collection.json"
-    },
-    "item": [
-        {
-            "name": "System",
-            "description": "Health checks, server metadata, supported formats, and spaCy NER labels.",
-            "item": [
-                {
-                    "name": "Root Health (No Auth)",
-                    "request": {
-                        "method": "GET",
-                        "header": [],
-                        "url": {
-                            "raw": "{{base_url}}/health",
-                            "host": [
-                                "{{base_url}}"
-                            ],
-                            "path": [
-                                "health"
-                            ]
-                        },
-                        "description": "Quick health check without authentication."
-                    },
-                    "response": []
-                },
-                {
-                    "name": "Ping (No Auth)",
-                    "request": {
-                        "method": "GET",
-                        "header": [],
-                        "url": {
-                            "raw": "{{base_url}}/ping",
-                            "host": [
-                                "{{base_url}}"
-                            ],
-                            "path": [
-                                "ping"
-                            ]
-                        },
-                        "description": "Simple ping endpoint without authentication."
-                    },
-                    "response": []
-                },
-                {
-                    "name": "Health",
-                    "request": {
-                        "method": "GET",
-                        "header": [
-                            {
-                                "key": "x-api-key",
-                                "value": "{{api_key}}",
-                                "type": "text"
-                            }
-                        ],
-                        "url": {
-                            "raw": "{{base_url}}/v1/health",
-                            "host": [
-                                "{{base_url}}"
-                            ],
-                            "path": [
-                                "v1",
-                                "health"
-                            ]
-                        },
-                        "description": "Authenticated health check."
-                    },
-                    "response": []
-                },
-                {
-                    "name": "Info",
-                    "request": {
-                        "method": "GET",
-                        "header": [
-                            {
-                                "key": "x-api-key",
-                                "value": "{{api_key}}",
-                                "type": "text"
-                            }
-                        ],
-                        "url": {
-                            "raw": "{{base_url}}/v1/info",
-                            "host": [
-                                "{{base_url}}"
-                            ],
-                            "path": [
-                                "v1",
-                                "info"
-                            ]
-                        },
-                        "description": "Server and environment information."
-                    },
-                    "response": []
-                },
-                {
-                    "name": "Supported Formats",
-                    "request": {
-                        "method": "GET",
-                        "header": [
-                            {
-                                "key": "x-api-key",
-                                "value": "{{api_key}}",
-                                "type": "text"
-                            }
-                        ],
-                        "url": {
-                            "raw": "{{base_url}}/v1/formats",
-                            "host": [
-                                "{{base_url}}"
-                            ],
-                            "path": [
-                                "v1",
-                                "formats"
-                            ]
-                        },
-                        "description": "List all supported file formats."
-                    },
-                    "response": []
-                },
-                {
-                    "name": "spaCy Labels",
-                    "request": {
-                        "method": "GET",
-                        "header": [
-                            {
-                                "key": "x-api-key",
-                                "value": "{{api_key}}",
-                                "type": "text"
-                            }
-                        ],
-                        "url": {
-                            "raw": "{{base_url}}/v1/spacy-labels",
-                            "host": [
-                                "{{base_url}}"
-                            ],
-                            "path": [
-                                "v1",
-                                "spacy-labels"
-                            ]
-                        },
-                        "description": "Available spaCy NER labels and mappings."
-                    },
-                    "response": []
-                }
-            ]
-        },
-        {
-            "name": "Convert",
-            "description": "Single-file and single-URL conversion to Markdown with optional structured JSON extraction.",
-            "item": [
-                {
-                    "name": "Convert File",
-                    "request": {
-                        "method": "POST",
-                        "header": [
-                            {
-                                "key": "x-api-key",
-                                "value": "{{api_key}}",
-                                "type": "text"
-                            }
-                        ],
-                        "body": {
-                            "mode": "formdata",
-                            "formdata": [
-                                {
-                                    "key": "file",
-                                    "type": "file",
-                                    "src": "/path/to/your/document.pdf",
-                                    "description": "The file to convert"
-                                },
-                                {
-                                    "key": "plain_text",
-                                    "value": "false",
-                                    "type": "text"
-                                },
-                                {
-                                    "key": "return_json",
-                                    "value": "false",
-                                    "type": "text"
-                                },
-                                {
-                                    "key": "mappings",
-                                    "value": "{\"company\":{\"source_type\":\"entity\",\"label\":\"ORG\"}}",
-                                    "type": "text"
-                                }
-                            ]
-                        },
-                        "url": {
-                            "raw": "{{base_url}}/v1/convert/file",
-                            "host": [
-                                "{{base_url}}"
-                            ],
-                            "path": [
-                                "v1",
-                                "convert",
-                                "file"
-                            ]
-                        },
-                        "description": "Upload a single file and convert it to Markdown."
-                    },
-                    "response": []
-                },
-                {
-                    "name": "Convert URL",
-                    "request": {
-                        "method": "POST",
-                        "header": [
-                            {
-                                "key": "x-api-key",
-                                "value": "{{api_key}}",
-                                "type": "text"
-                            },
-                            {
-                                "key": "Content-Type",
-                                "value": "application/json",
-                                "type": "text"
-                            }
-                        ],
-                        "body": {
-                            "mode": "raw",
-                            "raw": "{\n  \"url\": \"https://example.com/document.pdf\",\n  \"return_json\": false,\n  \"mappings\": {\n    \"company\": {\n      \"source_type\": \"entity\",\n      \"label\": \"ORG\"\n    }\n  }\n}",
-                            "options": {
-                                "raw": {
-                                    "language": "json"
-                                }
-                            }
-                        },
-                        "url": {
-                            "raw": "{{base_url}}/v1/convert/url",
-                            "host": [
-                                "{{base_url}}"
-                            ],
-                            "path": [
-                                "v1",
-                                "convert",
-                                "url"
-                            ]
-                        },
-                        "description": "Convert a remote document URL to Markdown."
-                    },
-                    "response": []
-                }
-            ]
-        },
-        {
-            "name": "Batch",
-            "description": "Bulk conversion of multiple files and URLs.",
-            "item": [
-                {
-                    "name": "Batch Files",
-                    "request": {
-                        "method": "POST",
-                        "header": [
-                            {
-                                "key": "x-api-key",
-                                "value": "{{api_key}}",
-                                "type": "text"
-                            }
-                        ],
-                        "body": {
-                            "mode": "formdata",
-                            "formdata": [
-                                {
-                                    "key": "files",
-                                    "type": "file",
-                                    "src": "/path/to/document1.pdf"
-                                },
-                                {
-                                    "key": "files",
-                                    "type": "file",
-                                    "src": "/path/to/document2.docx"
-                                }
-                            ]
-                        },
-                        "url": {
-                            "raw": "{{base_url}}/v1/batch/files",
-                            "host": [
-                                "{{base_url}}"
-                            ],
-                            "path": [
-                                "v1",
-                                "batch",
-                                "files"
-                            ]
-                        },
-                        "description": "Upload and convert multiple files."
-                    },
-                    "response": []
-                },
-                {
-                    "name": "Batch URLs",
-                    "request": {
-                        "method": "POST",
-                        "header": [
-                            {
-                                "key": "x-api-key",
-                                "value": "{{api_key}}",
-                                "type": "text"
-                            },
-                            {
-                                "key": "Content-Type",
-                                "value": "application/json",
-                                "type": "text"
-                            }
-                        ],
-                        "body": {
-                            "mode": "raw",
-                            "raw": "{\n  \"urls\": [\n    \"https://example.com/report1.pdf\",\n    \"https://example.com/report2.docx\"\n  ]\n}",
-                            "options": {
-                                "raw": {
-                                    "language": "json"
-                                }
-                            }
-                        },
-                        "url": {
-                            "raw": "{{base_url}}/v1/batch/urls",
-                            "host": [
-                                "{{base_url}}"
-                            ],
-                            "path": [
-                                "v1",
-                                "batch",
-                                "urls"
-                            ]
-                        },
-                        "description": "Convert multiple URLs."
-                    },
-                    "response": []
-                }
-            ]
-        }
-    ],
-    "event": [
-        {
-            "listen": "prerequest",
-            "script": {
-                "type": "text/javascript",
-                "exec": [
-                    "// Validate required collection variables before each request",
-                    "const baseUrl = pm.collectionVariables.get('base_url');",
-                    "const apiKey = pm.collectionVariables.get('api_key');",
-                    "",
-                    "if (!baseUrl) {",
-                    "    pm.expect.fail('Missing collection variable: base_url');",
-                    "}",
-                    "",
-                    "if (!apiKey && pm.request.url.path.join('/').startsWith('v1')) {",
-                    "    pm.expect.fail('Missing collection variable: api_key');",
-                    "}"
-                ]
-            }
-        }
-    ],
-    "variable": [
-        {
-            "key": "base_url",
-            "value": "http://localhost:7860",
-            "type": "string",
-            "description": "Base URL of the service"
-        },
-        {
-            "key": "api_key",
-            "value": "changeme",
-            "type": "string",
-            "description": "API key for authenticated endpoints"
-        }
-    ],
-    "auth": {
-        "type": "apikey",
-        "apikey": [
-            {
-                "key": "value",
-                "value": "{{api_key}}",
-                "type": "string"
-            },
-            {
-                "key": "key",
-                "value": "x-api-key",
-                "type": "string"
-            }
-        ]
-    }
-}

pyproject.toml

@@ -3,9 +3,9 @@ requires = ["setuptools>=70", "wheel"]
 build-backend = "setuptools.build_meta"
 
 [project]
-name = "reconciliation-non-ai-extractor"
+name = "All API Collection"
 version = "1.0.0"
-description = "Enterprise document extraction API powered by Microsoft MarkItDown, RapidOCR, and spaCy"
+description = "All API Collection - Document extraction, conversion, and database query API"
 readme = "README.md"
 requires-python = ">=3.10"
 license = { text = "MIT" }
@@ -34,4 +34,4 @@ dev = ["pytest>=8", "pytest-asyncio>=0.23"]
 
 [tool.setuptools.packages.find]
 where = ["."]
-include = ["reconciliation_non_ai_extractor*"]
+include = ["All API Collection*"]

requirements.txt

@@ -12,3 +12,8 @@ pillow>=10.0.0
 pypdfium2>=4.30.0
 pandas>=2.0.0
 spacy>=3.7.0
+
+# Async database drivers
+aiomysql>=0.3.2
+asyncpg>=0.31.0
+motor>=3.7.1