feat(python-driver): add public API for connection pooling and model dict conversion

drivers/python/age/__init__.py

@@ -16,6 +16,7 @@
 import psycopg.conninfo as conninfo
 from . import age
 from .age import *
+from .age import AgeLoader, ClientCursor, configure_connection
 from .models import *
 from .builder import ResultHandler, DummyResultHandler, parseAgeValue, newResultHandler
 from . import VERSION 

drivers/python/age/age.py

@@ -14,6 +14,8 @@
 # under the License.
 
 import re
+from typing import Any, Optional
+
 import psycopg
 from psycopg.types import TypeInfo
 from psycopg import sql
@@ -170,6 +172,71 @@ def setUpAge(conn:psycopg.connection, graphName:str, load_from_plugins:bool=Fals
         if graphName != None:
             checkGraphCreated(conn, graphName)
 
+
+def configure_connection(
+    conn: psycopg.connection,
+    graph_name: Optional[str] = None,
+    load: bool = False,
+    load_from_plugins: bool = False,
+) -> None:
+    """Register AGE agtype adapters on an existing connection.
+
+    This enables use of AGE with externally-managed connections, such as
+    those from psycopg_pool.ConnectionPool.  By default the function does
+    **not** execute ``LOAD 'age'``, making it safe for managed PostgreSQL
+    services (Azure, AWS RDS) where the extension is pre-loaded via
+    ``shared_preload_libraries``.
+
+    Performs:
+    - ``SET search_path`` to include ``ag_catalog``
+    - Fetches agtype OIDs and registers ``AgeLoader``
+    - Optionally loads the AGE extension (``load=True``)
+    - Optionally checks/creates the graph
+
+    Args:
+        conn: An existing psycopg connection.
+        graph_name: Optional graph name to check/create.
+        load: If True, execute ``LOAD 'age'`` (or the plugins path).
+            Default False — suitable for environments where AGE is
+            already loaded.
+        load_from_plugins: If True (and ``load=True``), use
+            ``LOAD '$libdir/plugins/age'`` instead of ``LOAD 'age'``.
+
+    Raises:
+        ValueError: If ``load_from_plugins=True`` but ``load=False``.
+        AgeNotSet: If the agtype type is not found in the database.
+    """
+    if load_from_plugins and not load:
+        raise ValueError(
+            "load_from_plugins=True requires load=True. "
+            "Set load=True to enable extension loading."
+        )
+
+    with conn.cursor() as cursor:
+        if load:
+            if load_from_plugins:
+                cursor.execute("LOAD '$libdir/plugins/age';")
+            else:
+                cursor.execute("LOAD 'age';")
+
+        cursor.execute('SET search_path = ag_catalog, "$user", public;')
+
+        ag_info = TypeInfo.fetch(conn, 'agtype')
+
+    if not ag_info:
+        raise AgeNotSet(
+            "AGE agtype type not found. Ensure the AGE extension is "
+            "installed and loaded in the current database. "
+            "Run CREATE EXTENSION age; first."
+        )
+
+    conn.adapters.register_loader(ag_info.oid, AgeLoader)
+    conn.adapters.register_loader(ag_info.array_oid, AgeLoader)
+
+    if graph_name is not None:
+        checkGraphCreated(conn, graph_name)
+
+
 # Create the graph, if it does not exist
 def checkGraphCreated(conn:psycopg.connection, graphName:str):
     validate_graph_name(graphName)

drivers/python/age/models.py

@@ -118,6 +118,20 @@ def toJson(self) -> str:
 
         return buf.getvalue()
 
+    def to_dict(self) -> list:
+        # AGObj elements are recursively converted; JSON-native types
+        # (dict, list, str, int, float, bool, None) pass through unchanged.
+        # Non-serializable objects fall back to str() as a safety net.
+        result = []
+        for e in (self.entities or []):
+            if isinstance(e, AGObj):
+                result.append(e.to_dict())
+            elif isinstance(e, (dict, list, str, int, float, bool, type(None))):
+                result.append(e)
+            else:
+                result.append(str(e))
+        return result
+
 
 
 
@@ -146,6 +160,18 @@ def __str__(self) -> str:
     def __repr__(self) -> str:
         return self.toString()
 
+    def to_dict(self) -> dict:
+        """Return a plain dict suitable for JSON serialization.
+
+        Properties are shallow-copied; nested mutable values will share
+        references with the original Vertex.
+        """
+        return {
+            "id": self.id,
+            "label": self.label,
+            "properties": dict(self.properties) if self.properties else {},
+        }
+
     def toString(self) -> str: 
         return nodeToString(self)
 
@@ -186,6 +212,20 @@ def __str__(self) -> str:
     def __repr__(self) -> str:
         return self.toString()
 
+    def to_dict(self) -> dict:
+        """Return a plain dict suitable for JSON serialization.
+
+        Properties are shallow-copied; nested mutable values will share
+        references with the original Edge.
+        """
+        return {
+            "id": self.id,
+            "label": self.label,
+            "start_id": self.start_id,
+            "end_id": self.end_id,
+            "properties": dict(self.properties) if self.properties else {},
+        }
+
     def extraStrFormat(node, buf):
         if node.start_id != None:
             buf.write(", start_id:")