Async client for use with Python's asynchronous I/O + examples (#14)

* Async client for use with Python's asynchronous I/O + examples

Here we add support for Python's built-in asyncio [1], a feature that's
commonly requested for library APIs because it enables very performant
code built on asynchronous I/O.

The change creates a second client class called `AsyncClient` along with
an `riversqlalchemy.AsyncDriver` to go with the original driver. This
design is based on SQLAlchemy's and the code that the Python sqlc plugin
generates. The APIs for the pair of clients and drivers are identical,
but one set has `async` annotations on everything and takes appropriate
inputs (like an async SQLAlchemy engine instead of synchronous one).

The final usage looks pretty close to non-async code (which I think is a
pretty reasonable outcome overall):

    engine = sqlalchemy.ext.asyncio.create_async_engine(dev_database_url(is_async=True))
    client = riverqueue.AsyncClient(riversqlalchemy.AsyncDriver(engine))

    insert_res = await client.insert(
        SortArgs(strings=["whale", "tiger", "bear"]),
        insert_opts=riverqueue.InsertOpts(
            unique_opts=riverqueue.UniqueOpts(by_period=900)
        ),
    )
    print(insert_res)

I also added a new `examples/` package that shows code for how to do
various things with River Python, and makes sure they're run in CI and
get static analysis from MyPy.

[1] https://docs.python.org/3/library/asyncio.html

* Drop Docker-based tests in favor of normal Postgres

Drop Docker-based tests (or at least within the Python program) in favor
of using normal Postgres like the other River projects are doing. I'm
having a lot of trouble getting Docker working in conjunction with
async. It's probably possible to get it working, but the testcontainers
docs are subpar, and I'm having to iterate failing tests in GitHub
Actions which is awful.

Author: brandur

.github/workflows/ci.yaml

@@ -1,23 +1,56 @@
-name: build
+name: CI
 
 on:
   push:
     branches:
       - master
   pull_request:
 
+env:
+  # Database to connect to that can create other databases with `CREATE DATABASE`.
+  ADMIN_DATABASE_URL: postgres://postgres:postgres@localhost:5432
+
+  # A suitable name/URL for non-test database.
+  DATABASE_NAME: river_dev
+  DATABASE_URL: postgres://postgres:postgres@127.0.0.1:5432/river_dev
+
+  # Test database.
+  TEST_DATABASE_NAME: river_test
+  TEST_DATABASE_URL: postgres://postgres:postgres@127.0.0.1:5432/river_test
+
 jobs:
-  build:
+  build_and_test:
     runs-on: ubuntu-latest
 
+    services:
+      postgres:
+        image: postgres:latest
+        env:
+          POSTGRES_PASSWORD: postgres
+        options: >-
+          --health-cmd pg_isready
+          --health-interval 2s
+          --health-timeout 5s
+          --health-retries 5
+        ports:
+          - 5432:5432
+
     steps:
       - name: Checkout
         uses: actions/checkout@v4
 
-      - name: Install the latest version of rye
+      - name: Install Rye
         uses: eifinger/setup-rye@v3
 
-      - name: Rye setup
+      # Needed for River's CLI. There is a version of Go on Actions' base image,
+      # but it's old and can't read modern `go.mod` annotations correctly.
+      - name: Install Go
+        uses: actions/setup-go@v5
+        with:
+          go-version: "stable"
+          check-latest: true
+
+      - name: Rye sync
         run: rye sync
 
       - name: Format check
@@ -32,5 +65,60 @@ jobs:
       - name: Build
         run: rye build
 
+      - name: Install River CLI
+        run: go install github.com/riverqueue/river/cmd/river@latest
+
+      - name: Create database
+        run: psql --echo-errors --quiet -c '\timing off' -c "CREATE DATABASE ${TEST_DATABASE_NAME};" ${ADMIN_DATABASE_URL}
+
+      - name: river migrate-up
+        run: river migrate-up --database-url "$TEST_DATABASE_URL"
+
       - name: Test
-        run: RIVER_USE_DOCKER=true rye test
+        run: rye test
+
+  examples_run:
+    runs-on: ubuntu-latest
+
+    services:
+      postgres:
+        image: postgres:latest
+        env:
+          POSTGRES_PASSWORD: postgres
+        options: >-
+          --health-cmd pg_isready
+          --health-interval 2s
+          --health-timeout 5s
+          --health-retries 5
+        ports:
+          - 5432:5432
+
+    steps:
+      - name: Checkout
+        uses: actions/checkout@v4
+
+      - name: Install Rye
+        uses: eifinger/setup-rye@v3
+
+      # Needed for River's CLI. There is a version of Go on Actions' base image,
+      # but it's old and can't read modern `go.mod` annotations correctly.
+      - name: Install Go
+        uses: actions/setup-go@v5
+        with:
+          go-version: "stable"
+          check-latest: true
+
+      - name: Rye sync
+        run: rye sync
+
+      - name: Install River CLI
+        run: go install github.com/riverqueue/river/cmd/river@latest
+
+      - name: Create database
+        run: psql --echo-errors --quiet -c '\timing off' -c "CREATE DATABASE ${DATABASE_NAME};" ${ADMIN_DATABASE_URL}
+
+      - name: river migrate-up
+        run: river migrate-up --database-url "$DATABASE_URL"
+
+      - name: Run examples
+        run: rye run python3 -m examples.all

Makefile

@@ -20,4 +20,4 @@ test: ## Run test suite with Rye/pytest
 
 .PHONY: type-check
 type-check: ## Run type check with MyPy
-	rye run mypy -p riverqueue -p tests
+	rye run mypy -p riverqueue -p examples -p tests

docs/development.md

@@ -36,12 +36,6 @@ Run all tests:
 $ rye test
 ```
 
-_Or_, using a Docker test Postgres container instead of a test database:
-
-```shell
-RIVER_USE_DOCKER=true rye test
-```
-
 Run a specific test (or without `-k` option for all tests in a single file):
 
 ```shell

examples/__init__.py

@@ -0,0 +1,25 @@
+#
+# Run with:
+#
+#     rye run python3 -m examples.all
+#
+# It'd be nice if this could pick up every example automatically, but after
+# spending an hour just trying to find a way of making a relative import work,
+# I'm not going anywhere near that problem right now.
+#
+
+import asyncio
+
+from examples import async_client_insert_example
+from examples import async_client_insert_tx_example
+from examples import client_insert_example
+from examples import client_insert_tx_example
+
+if __name__ == "__main__":
+    asyncio.set_event_loop(asyncio.new_event_loop())
+
+    asyncio.run(async_client_insert_example.example())
+    asyncio.run(async_client_insert_tx_example.example())
+
+    client_insert_example.example()
+    client_insert_tx_example.example()

examples/all.py

@@ -0,0 +1,42 @@
+#
+# Run with:
+#
+#     rye run python3 -m examples.async_client_insert_example
+#
+
+import asyncio
+from dataclasses import dataclass
+import json
+import riverqueue
+import sqlalchemy
+
+from examples.helpers import dev_database_url
+from riverqueue.driver import riversqlalchemy
+
+
+@dataclass
+class SortArgs:
+    strings: list[str]
+
+    kind: str = "sort"
+
+    def to_json(self) -> str:
+        return json.dumps({"strings": self.strings})
+
+
+async def example():
+    engine = sqlalchemy.ext.asyncio.create_async_engine(dev_database_url(is_async=True))
+    client = riverqueue.AsyncClient(riversqlalchemy.AsyncDriver(engine))
+
+    insert_res = await client.insert(
+        SortArgs(strings=["whale", "tiger", "bear"]),
+        insert_opts=riverqueue.InsertOpts(
+            unique_opts=riverqueue.UniqueOpts(by_period=900)
+        ),
+    )
+    print(insert_res)
+
+
+if __name__ == "__main__":
+    asyncio.set_event_loop(asyncio.new_event_loop())
+    asyncio.run(example())

examples/async_client_insert_example.py

@@ -0,0 +1,44 @@
+#
+# Run with:
+#
+#     rye run python3 -m examples.async_client_insert_tx_example
+#
+
+import asyncio
+from dataclasses import dataclass
+import json
+import riverqueue
+import sqlalchemy
+
+from examples.helpers import dev_database_url
+from riverqueue.driver import riversqlalchemy
+
+
+@dataclass
+class SortArgs:
+    strings: list[str]
+
+    kind: str = "sort"
+
+    def to_json(self) -> str:
+        return json.dumps({"strings": self.strings})
+
+
+async def example():
+    engine = sqlalchemy.ext.asyncio.create_async_engine(dev_database_url(is_async=True))
+    client = riverqueue.AsyncClient(riversqlalchemy.AsyncDriver(engine))
+
+    async with engine.begin() as session:
+        insert_res = await client.insert_tx(
+            session,
+            SortArgs(strings=["whale", "tiger", "bear"]),
+            insert_opts=riverqueue.InsertOpts(
+                unique_opts=riverqueue.UniqueOpts(by_period=900)
+            ),
+        )
+        print(insert_res)
+
+
+if __name__ == "__main__":
+    asyncio.set_event_loop(asyncio.new_event_loop())
+    asyncio.run(example())

examples/async_client_insert_tx_example.py

@@ -0,0 +1,40 @@
+#
+# Run with:
+#
+#     rye run python3 -m examples.client_insert_example
+#
+
+from dataclasses import dataclass
+import json
+import riverqueue
+import sqlalchemy
+
+from examples.helpers import dev_database_url
+from riverqueue.driver import riversqlalchemy
+
+
+@dataclass
+class SortArgs:
+    strings: list[str]
+
+    kind: str = "sort"
+
+    def to_json(self) -> str:
+        return json.dumps({"strings": self.strings})
+
+
+def example():
+    engine = sqlalchemy.create_engine(dev_database_url())
+    client = riverqueue.Client(riversqlalchemy.Driver(engine))
+
+    insert_res = client.insert(
+        SortArgs(strings=["whale", "tiger", "bear"]),
+        insert_opts=riverqueue.InsertOpts(
+            unique_opts=riverqueue.UniqueOpts(by_period=900)
+        ),
+    )
+    print(insert_res)
+
+
+if __name__ == "__main__":
+    example()

examples/client_insert_example.py

@@ -0,0 +1,42 @@
+#
+# Run with:
+#
+#     rye run python3 -m examples.client_insert_tx_example
+#
+
+from dataclasses import dataclass
+import json
+import riverqueue
+import sqlalchemy
+
+from examples.helpers import dev_database_url
+from riverqueue.driver import riversqlalchemy
+
+
+@dataclass
+class SortArgs:
+    strings: list[str]
+
+    kind: str = "sort"
+
+    def to_json(self) -> str:
+        return json.dumps({"strings": self.strings})
+
+
+def example():
+    engine = sqlalchemy.create_engine(dev_database_url())
+    client = riverqueue.Client(riversqlalchemy.Driver(engine))
+
+    with engine.begin() as session:
+        insert_res = client.insert_tx(
+            session,
+            SortArgs(strings=["whale", "tiger", "bear"]),
+            insert_opts=riverqueue.InsertOpts(
+                unique_opts=riverqueue.UniqueOpts(by_period=900)
+            ),
+        )
+        print(insert_res)
+
+
+if __name__ == "__main__":
+    example()

examples/client_insert_tx_example.py

@@ -0,0 +1,14 @@
+import os
+
+
+def dev_database_url(is_async: bool = False) -> str:
+    database_url = os.getenv("DATABASE_URL", "postgres://localhost/river_dev")
+
+    # sqlalchemy removed support for postgres:// for reasons beyond comprehension
+    database_url = database_url.replace("postgres://", "postgresql://")
+
+    # mix in an async driver for async
+    if is_async:
+        database_url = database_url.replace("postgresql://", "postgresql+asyncpg://")
+
+    return database_url