feat(postgres/sqlite): update API to be fully async

Signed-off-by: Vaughn Dice <vdice@akamai.com>

Author: vdice

examples/spin-postgres/app.py

@@ -1,14 +1,12 @@
-from spin_sdk import http, postgres
+from spin_sdk import http, postgres, util
 from spin_sdk.http import Request, Response
-from spin_sdk.postgres import RowSet, DbValue
-
+from spin_sdk.postgres import Connection, RowSet, DbValue
 
 def format_value(db_value: DbValue) -> str:
     if hasattr(db_value, "value"):
         return str(db_value.value)
     return "NULL"
 
-
 def format_rowset(rowset: RowSet) -> str:
     lines = []
     col_names = [col.name for col in rowset.columns]
@@ -19,11 +17,12 @@ def format_rowset(rowset: RowSet) -> str:
         lines.append(" | ".join(values))
     return "\n".join(lines)
 
-
 class HttpHandler(http.Handler):
     async def handle_request(self, request: Request) -> Response:
-        with await postgres.open("user=postgres dbname=spin_dev host=localhost sslmode=disable password=password") as db:
-            rowset = db.query("SELECT * FROM test", [])
+        with await Connection.open("user=postgres dbname=spin_dev host=localhost sslmode=disable password=password") as db:
+            columns, stream, future = await db.query("SELECT * FROM test", [])
+            rows = await util.collect((stream, future))
+            rowset = RowSet(columns, list(rows))
             result = format_rowset(rowset)
 
         return Response(200, {"content-type": "text/plain"}, bytes(result, "utf-8"))

examples/spin-sqlite/app.py

@@ -1,13 +1,14 @@
-from spin_sdk import http, sqlite
+from spin_sdk import http, util
 from spin_sdk.http import Request, Response
-from spin_sdk.sqlite import Value_Integer
+from spin_sdk.sqlite import Connection, Value_Text, Value_Integer
 
 class HttpHandler(http.Handler):
     async def handle_request(self, request: Request) -> Response:
-        with await sqlite.open_default() as db:
-            result = db.execute("SELECT * FROM todos WHERE id > (?);", [Value_Integer(1)])
-            rows = result.rows
-        
+        with await Connection.open_default() as db:
+            await db.execute("INSERT INTO todos (description, due) VALUES (?, ?)", [Value_Text("Try out Spin SQLite"), Value_Text("Friday")])
+            columns, stream, future = await db.execute("SELECT * FROM todos WHERE id > (?);", [Value_Integer(1)])
+            rows = await util.collect((stream, future))
+
         return Response(
             200,
             {"content-type": "text/plain"},

src/spin_sdk/postgres.py

@@ -1,19 +1,58 @@
 """Module for interacting with a Postgres database"""
 
+from typing import List, Tuple
+from types import TracebackType
+from componentize_py_types import Result
+from componentize_py_async_support.streams import StreamReader
+from componentize_py_async_support.futures import FutureReader
 from spin_sdk.wit.imports import spin_postgres_postgres_4_2_0 as pg
 
-Connection = pg.Connection
 RowSet = pg.RowSet
 DbValue = pg.DbValue
+ParameterValue = pg.ParameterValue
+Column = pg.Column
+Error  = pg.Error
 
-async def open(connection_string: str) -> Connection:
-    """
-    Open a connection with a Postgres database.
-    
-    The connection_string is the Postgres URL connection string.
-
-    A `componentize_py_types.Err(Error_ConnectionFailed(str))` when a connection fails.
-    
-    A `componentize_py_types.Err(Error_Other(str))` when some other error occurs.
-    """
-    return await Connection.open_async(connection_string)
+class Connection:
+    def __init__(self, connection: pg.Connection) -> None:
+        self.connection = connection
+
+    def __enter__(self) -> Connection:
+        return self
+
+    def __exit__(self, exc_type: type[BaseException] | None, exc_value: BaseException | None, traceback: TracebackType | None) -> bool | None:
+        return self.connection.__exit__(exc_type, exc_value, traceback)
+
+    @classmethod
+    async def open(cls, connection_string: str) -> Connection:
+        """
+        Open a connection with a Postgres database.
+
+        The connection_string is the Postgres URL connection string.
+
+        A `componentize_py_types.Err(Error_ConnectionFailed(str))` when a connection fails.
+
+        A `componentize_py_types.Err(Error_Other(str))` when some other error occurs.
+        """
+        return Connection(await pg.Connection.open_async(connection_string))
+
+    async def query(self, statement: str, params: List[ParameterValue]) -> Tuple[List[Column], StreamReader[List[DbValue]], FutureReader[Result[None, Error]]]:
+        """
+        Query the Postgres database.
+
+        Returns a Tuple containing a List of Columns, a List of DbValues encapsulated in an asynchronous iterator (`componentize_py_async_support.streams.StreamReader`),
+        and a future (`componentize_py_async_support.futures.FutureReader`) containing the result of the operation.
+
+        Raises: `componentize_py_types.Err(spin_sdk.wit.imports.spin_postgres_postgres_4_2_0.Error)`
+        """
+        return await self.connection.query_async(statement, params)
+
+    async def execute(self, statement: str, params: List[ParameterValue]) -> int:
+        """
+        Execute command to the database.
+
+        Returns the number of affected rows as an int.
+
+        Raises: `componentize_py_types.Err(spin_sdk.wit.imports.spin_postgres_postgres_4_2_0.Error)`
+        """
+        return await self.connection.execute_async(statement, params)

src/spin_sdk/sqlite.py

@@ -1,34 +1,77 @@
 """Module for interacting with an SQLite database"""
 
-from typing import List
+from typing import List, Tuple
+from types import TracebackType
+from componentize_py_types import Result
+from componentize_py_async_support.streams import StreamReader
+from componentize_py_async_support.futures import FutureReader
 from spin_sdk.wit.imports import spin_sqlite_sqlite_3_1_0 as sqlite
 
-Connection = sqlite.Connection
+Error = sqlite.Error
+Value = sqlite.Value
 Value_Integer = sqlite.Value_Integer
 Value_Real = sqlite.Value_Real
 Value_Text = sqlite.Value_Text
 Value_Blob = sqlite.Value_Blob
+RowResult = sqlite.RowResult
 
-async def open(name: str) -> Connection:
-    """Open a connection to a named database instance.
+class Connection:
+    def __init__(self, connection: sqlite.Connection) -> None:
+        self.connection = connection
 
-    If `database` is "default", the default instance is opened.
+    def __enter__(self) -> Connection:
+        return self
 
-    A `componentize_py_types.Err(spin_sdk.wit.imports.spin_sqlite_sqlite_3_1_0.Error_AccessDenied)` will be raised when the component does not have access to the specified database.
+    def __exit__(self, exc_type: type[BaseException] | None, exc_value: BaseException | None, traceback: TracebackType | None) -> bool | None:
+        return self.connection.__exit__(exc_type, exc_value, traceback)
 
-    A `componentize_py_types.Err(spin_sdk.wit.imports.spin_sqlite_sqlite_3_1_0.Error_NoSuchDatabase)` will be raised when the host does not recognize the database name requested.
-    
-    A `componentize_py_types.Err(spin_sdk.wit.imports.spin_sqlite_sqlite_3_1_0.Error_InvalidConnection)` will be raised when the provided connection string is not valid.
-    
-    A `componentize_py_types.Err(spin_sdk.wit.imports.spin_sqlite_sqlite_3_1_0.Error_Io(str))` will be raised when implementation-specific error occured (e.g. I/O)
-    """
-    return await Connection.open_async(name)
+    @classmethod
+    async def open(cls, name: str) -> Connection:
+        """Open a connection to a named database instance.
 
-async def open_default() -> Connection:
-    """Open the default store.
+        If `database` is "default", the default instance is opened.
 
-    A `componentize_py_types.Err(spin_sdk.wit.imports.spin_sqlite_sqlite_3_1_0.Error_AccessDenied)` will be raised when the component does not have access to the default database.
+        A `componentize_py_types.Err(spin_sdk.wit.imports.spin_sqlite_sqlite_3_1_0.Error_AccessDenied)` will be raised when the component does not have access to the specified database.
 
-    A `componentize_py_types.Err(spin_sdk.wit.imports.spin_sqlite_sqlite_3_1_0.Error_Io(str))` will be raised when implementation-specific error occured (e.g. I/O)
-    """
-    return await Connection.open_async("default")
+        A `componentize_py_types.Err(spin_sdk.wit.imports.spin_sqlite_sqlite_3_1_0.Error_NoSuchDatabase)` will be raised when the host does not recognize the database name requested.
+
+        A `componentize_py_types.Err(spin_sdk.wit.imports.spin_sqlite_sqlite_3_1_0.Error_InvalidConnection)` will be raised when the provided connection string is not valid.
+
+        A `componentize_py_types.Err(spin_sdk.wit.imports.spin_sqlite_sqlite_3_1_0.Error_Io(str))` will be raised when implementation-specific error occured (e.g. I/O)
+        """
+        return Connection(await sqlite.Connection.open_async(name))
+
+    @classmethod
+    async def open_default(cls) -> Connection:
+        """Open the default store.
+
+        A `componentize_py_types.Err(spin_sdk.wit.imports.spin_sqlite_sqlite_3_1_0.Error_AccessDenied)` will be raised when the component does not have access to the default database.
+
+        A `componentize_py_types.Err(spin_sdk.wit.imports.spin_sqlite_sqlite_3_1_0.Error_Io(str))` will be raised when implementation-specific error occured (e.g. I/O)
+        """
+        return Connection(await sqlite.Connection.open_async("default"))
+
+    async def execute(self, statement: str, params: List[Value]) -> Tuple[List[str], StreamReader[RowResult], FutureReader[Result[None, Error]]]:
+        """
+        Execute a command to the database.
+
+        Returns a Tuple containing a List of columns, a List of RowResults encapsulated in an asynchronous iterator (`componentize_py_async_support.streams.StreamReader`),
+        and a future (`componentize_py_async_support.futures.FutureReader`) containing the result of the operation.
+
+        Raises: `componentize_py_types.Err(spin_sdk.wit.imports.spin_sqlite_sqlite_3_1_0.Error)`
+        """
+        return await self.connection.execute_async(statement, params)
+
+    async def last_insert_rowid(self) -> int:
+        """
+        The SQLite rowid of the most recent successful INSERT on the connection, or 0 if
+        there has not yet been an INSERT on the connection.
+        """
+        return await self.connection.last_insert_rowid_async()
+
+    async def changes(self) -> int:
+        """
+        The number of rows modified, inserted or deleted by the most recently completed
+        INSERT, UPDATE or DELETE statement on the connection.
+        """
+        return await self.connection.changes_async()