$j/dZddlZddlmZmZddlmZddlmZddl m Z m Z ddl m Z e rdd lmZGd d eZGd d eZdS)zAbstract base classes for database backend abstraction. Defines the interfaces that all database backends (PostgreSQL, Oracle, etc.) must implement. Business logic depends only on these interfaces. N)ABCabstractmethod) AsyncIterator)asynccontextmanager) TYPE_CHECKINGAny) ResultRow) DataAccessOpscheZdZdZedefdZdedefdZdddded e ed e e d e edzd edzde e ezf d Z e e dedfdZe dddedededzdefdZe dddede eedfdedzddfdZe dddedededzde e fdZe dddedededzde dzfdZe ddddededededzdef dZdddede eedfd e ededzddf dZdS)DatabaseConnectionzWraps a single connection from the pool. Provides a uniform interface over asyncpg.Connection, oracledb cursor, etc. Methods mirror asyncpg's connection API for minimal migration friction. returncdSz(Return ``"postgresql"`` or ``"oracle"``. postgresqlselfs d/home/rurouni/.hermes/hermes-agent/venv/lib/python3.11/site-packages/hindsight_api/engine/db/base.py backend_typezDatabaseConnection.backend_type |valuec|dSt|tr3 tj|S#tjt f$r|cYSwxYw|S)aParse a JSON column value into a Python object. PG (asyncpg) returns JSON columns as strings that need json.loads(). Oracle returns them as pre-parsed dicts/lists (via OracleConnection row conversion). This method normalizes both to Python objects. N) isinstancestrjsonloadsJSONDecodeError TypeError)rrs r parse_jsonzDatabaseConnection.parse_json#sh =4 eS ! !  z%(((()4      s/A  A N) column_types returningtablecolumnsarraysr"r#c< Kd|}t|}|pdg|z d fdt|D}d|d|d|d} |r| d|z } |j| g|Rd {VS|j| g|Rd {V} | S) aInsert multiple rows from parallel arrays. Default implementation uses ``INSERT ... SELECT * FROM unnest(...)`` (PostgreSQL). Oracle overrides this with ``executemany``. Args: table: Fully-qualified table name. columns: Column names matching the arrays. arrays: Parallel lists of values, one per column. column_types: PG type suffixes for unnest casting (e.g. ``["text[]", "uuid[]"]``). Ignored by backends that don't use unnest. returning: Optional column expression for a RETURNING clause. Returns: If *returning* is set, a list of ResultRow; otherwise a status string. , ztext[]c3:K|]}d|dzd|VdS)$r z::Nr).0itypess r z=DatabaseConnection.bulk_insert_from_arrays..Qs<OOA 7AE 7 7U1X 7 7OOOOOOr INSERT INTO  (z) SELECT * FROM unnest()z RETURNING N)joinlenrangefetchexecute) rr$r%r&r"r#col_listn_cols unnest_argsqueryresultr-s @rbulk_insert_from_arraysz*DatabaseConnection.bulk_insert_from_arrays4s499W%%W3 V 3iiOOOOvOOOOO WuWWWWWWW  4 .9.. .E#E3F333333333 3#t|E3F333333333 rcK dWVdS)uStart a transaction (or savepoint if already in a transaction). Yields: Self — the same connection, now inside a transaction scope. On clean exit the transaction is committed; on exception it is rolled back. Nrrs r transactionzDatabaseConnection.transactionYs  r)timeoutr:argsr?c KdS)a;Execute a query and return a status string (e.g. 'INSERT 0 1'). Args: query: SQL query with dialect-appropriate placeholders. *args: Positional bind parameters. timeout: Optional statement timeout in seconds. Returns: Command status string. Nrrr:r?r@s rr6zDatabaseConnection.executee  r.c KdS)aExecute a query for each set of arguments. Args: query: SQL query with dialect-appropriate placeholders. args: List of argument tuples, one per execution. timeout: Optional statement timeout in seconds. Nr)rr:r@r?s r executemanyzDatabaseConnection.executemanyss  rc KdS)a$Execute a query and return all rows. Args: query: SQL query with dialect-appropriate placeholders. *args: Positional bind parameters. timeout: Optional statement timeout in seconds. Returns: List of ResultRow objects. NrrBs rr5zDatabaseConnection.fetch~rCrc KdS)aEExecute a query and return a single row (or None). Args: query: SQL query with dialect-appropriate placeholders. *args: Positional bind parameters. timeout: Optional statement timeout in seconds. Returns: A single ResultRow, or None if no rows match. NrrBs rfetchrowzDatabaseConnection.fetchrowrCrr)columnr?rIc KdS)aExecute a query and return a single value from the first row. Args: query: SQL query with dialect-appropriate placeholders. *args: Positional bind parameters. column: Column index to return (default 0). timeout: Optional statement timeout in seconds. Returns: The value from the specified column of the first row, or None. Nr)rr:rIr?r@s rfetchvalzDatabaseConnection.fetchvals  r table_namerecordscKd|}ddtt|D}d|d|d|d}||t |d{VdS)zBulk-load records into a table. Default implementation uses executemany INSERT. Backends with native bulk-load support (e.g. asyncpg COPY) should override for performance. r(c3&K|] }d|dzV dS)r*r Nr)r+r,s rr.z;DatabaseConnection.copy_records_to_table..s, J JQU J J J J J Jrr/r0z ) VALUES (r1N)r2r4r3rElist)rrLrMr%r?cols placeholdersr:s rcopy_records_to_tablez(DatabaseConnection.copy_records_to_tablesyy!!yy J JeCLL6I6I J J JJJ LzLLTLL\LLLud7mm44444444444r)__name__ __module__ __qualname____doc__propertyrrrr!rPr r<rrrr>floatr6tuplerEr5rHintrKrSrrrr r s3 cX.*. $####c#T # 3i$& #:# i3 ####J=1E#F^LP    3  s  UT\  UX    ^  dh   s $uS#X2G UZ]aUa mq   ^ JN      S  54<  SWXaSb    ^  MQ    C    edl  V_bfVf    ^  CD^b    C    S  uW[|  gj    ^  (!% 5555eCHo& 5 c 5  5 555555rr c,eZdZUdZdZded<edefdZed(dZ ede fd Z ede fd Z ede fd Z ede fd Zede fd ZdedzdedzfdZdddededzddfdZddddededefdZeddddddddedededed ed!ed"edzddfd#Zed)d$Zeedeefd%Zeedeefd&Zedefd'ZdS)*DatabaseBackenduDatabase pool lifecycle and connection acquisition. Manages the connection pool and provides context managers for acquiring connections and running transactions. The ``ops`` property provides backend-specific data access operations (the Strategy pattern — like Django's ``connection.ops``). All business logic should use ``backend.ops`` instead of creating DataAccessOps instances directly. NzDataAccessOps | None _ops_instancercdSrrrs rrzDatabaseBackend.backend_typerrr cT|jddlm}||j|_|jS)aBackend-specific data access operations (cached). Follows the Django pattern: ``connection.ops`` provides the operations handler for the current backend. Created lazily on first access and cached for the lifetime of the backend. Nr )create_data_access_ops)r^rar)rras ropszDatabaseBackend.opss>   % 0 0 0 0 0 0!7!78I!J!JD !!rcdS)u'Can CREATE INDEX … WHERE .Trrs rsupports_partial_indexesz(DatabaseBackend.supports_partial_indexes trcdS)z%Has BM25 / tsvector full-text search.Trrs r supports_bm25zDatabaseBackend.supports_bm25rfrcdS)z5Supports ``unnest()`` for expanding arrays into rows.Trrs rsupports_unnestzDatabaseBackend.supports_unnestrfrcdS)zAPlatform *might* have pg_trgm (must still be checked at runtime).Trrs rsupports_pg_trgmz DatabaseBackend.supports_pg_trgmrfrcdS)zWhether this backend supports the async WorkerPoller. WorkerPoller is backend-agnostic (uses DatabaseBackend.acquire()). All current backends (PostgreSQL, Oracle) support it. Trrs rsupports_worker_pollerz&DatabaseBackend.supports_worker_pollers trschemac|S)zNormalize a schema name for this backend. Returns the schema as-is by default. Oracle overrides this to convert ``"public"`` (a PG-specific default) to ``None`` (use the connecting user's default schema). r)rros rnormalize_schemaz DatabaseBackend.normalize_schemas  r)rodsncJtt|jd)zRun database migrations for this backend. PG uses Alembic migrations. Oracle uses its own idempotent DDL runner. Subclasses must override this method. z must implement run_migrations())NotImplementedErrortyperT)rrrros rrun_migrationszDatabaseBackend.run_migrationss$ "T$ZZ%8"Z"Z"Z[[[r pool_getter schema_getterrxryc(ddlm}|||S)zCreate the task backend for this database. All backends use BrokerTaskBackend for async worker/poller execution. )BrokerTaskBackendrw) task_backendr|)rrxryr|s rcreate_task_backendz#DatabaseBackend.create_task_backends, 544444  [ VVVVri,r)min_sizemax_sizecommand_timeoutacquire_timeoutstatement_cache_size init_callbackrrrrrrc KdS)aCreate the connection pool. Args: dsn: Database connection string. min_size: Minimum number of connections in the pool. max_size: Maximum number of connections in the pool. command_timeout: Default command timeout in seconds. acquire_timeout: Timeout for acquiring a connection from the pool. statement_cache_size: Size of the prepared-statement cache (0 to disable). init_callback: Optional async callback invoked on each new connection. Nr)rrrrrrrrrs r initializezDatabaseBackend.initializes . rc KdS)z4Close the connection pool and release all resources.Nrrs rshutdownzDatabaseBackend.shutdown1s  rcK dWVdS)zgAcquire a connection from the pool. Yields: A DatabaseConnection wrapper. Nrrs racquirezDatabaseBackend.acquire6s  rcK dWVdS)zAcquire a connection and start a transaction. The transaction is committed on clean exit, rolled back on exception. Yields: A DatabaseConnection wrapper inside a transaction. Nrrs rr>zDatabaseBackend.transactionAs  rcdS)uReturn the underlying raw pool object. Escape hatch for gradual migration — callers that still need direct pool access (e.g. asyncpg-specific features) can use this during the transition period. Nrrs rget_poolzDatabaseBackend.get_poolNs  r)rr )rN)rTrUrVrWr^__annotations__rXrrrcboolrerhrjrlrnrqrvrr~rr[rYrrrrr rr>rrrrr]r]s:  -1M)000 cX " " "X "$XtXX$XXsTzcDj@D\\\#\#*\\\\\9=SWWWW#WSW\_WWWW !$!#$%$(           " Tz     ^ 0   ^ }-?@^ =1C#D   ^  #   ^   rr])rWrabcrrcollections.abcr contextlibrtypingrrr;r rcr r r]rrrrs   ########))))))****** &%%%%%%%#""""""b5b5b5b5b5b5b5b5JZ Z Z Z Z cZ Z Z Z Z r