Manually edit python example code

The generated example code is edited just to demostrate potential
improvements to the python codegen.

The main changes are using sqlalchemy as the exection engine and using
stdlib dataclasses instead of pydantic. This also eliminates the runtime
package.

Author: Robert Holt

examples/python/requirements.txt

@@ -2,5 +2,4 @@ pytest~=6.2.2
 pytest-asyncio~=0.14.0
 psycopg2-binary~=2.8.6
 asyncpg~=0.21.0
-pydantic~=1.7.3
-sqlc-python-runtime~=1.0.0
+sqlalchemy==1.4.0b3

examples/python/src/authors/models.py

@@ -1,13 +1,13 @@
 # Code generated by sqlc. DO NOT EDIT.
 from typing import Optional
-
-import pydantic
+import dataclasses
 
 
 # Enums
 
 # Models
-class Author(pydantic.BaseModel):
+@dataclasses.dataclass()
+class Author:
     id: int
     name: str
     bio: Optional[str]

examples/python/src/authors/query.py

@@ -1,92 +1,77 @@
 # Code generated by sqlc. DO NOT EDIT.
-from typing import AsyncIterator, Awaitable, Iterator, Optional, overload
+from typing import AsyncIterator, Iterator, Optional
 
-import sqlc_runtime as sqlc
+import sqlalchemy
+import sqlalchemy.ext.asyncio
 
 from authors import models
 
 
-CREATE_AUTHOR = """-- name: create_author :one
+CREATE_AUTHOR = """-- name: create_author \\:one
 INSERT INTO authors (
           name, bio
 ) VALUES (
-  $1, $2
+  :p1, :p2
 )
 RETURNING id, name, bio
 """
 
 
-DELETE_AUTHOR = """-- name: delete_author :exec
+DELETE_AUTHOR = """-- name: delete_author \\:exec
 DELETE FROM authors
-WHERE id = $1
+WHERE id = :p1
 """
 
 
-GET_AUTHOR = """-- name: get_author :one
+GET_AUTHOR = """-- name: get_author \\:one
 SELECT id, name, bio FROM authors
-WHERE id = $1 LIMIT 1
+WHERE id = :p1 LIMIT 1
 """
 
 
-LIST_AUTHORS = """-- name: list_authors :many
+LIST_AUTHORS = """-- name: list_authors \\:many
 SELECT id, name, bio FROM authors
 ORDER BY name
 """
 
 
-@overload
-def create_author(conn: sqlc.Connection, name: str, bio: Optional[str]) -> Optional[models.Author]:
-    pass
+class Query:
+    def __init__(self, conn: sqlalchemy.engine.Connection):
+        self._conn = conn
 
+    def create_author(self, name: str, bio: Optional[str]) -> Optional[models.Author]:
+        result = self._conn.execute(sqlalchemy.text(CREATE_AUTHOR), {"p1": name, "p2": bio})
+        return models.Author(**dict(result.first()))
 
-@overload
-def create_author(conn: sqlc.AsyncConnection, name: str, bio: Optional[str]) -> Awaitable[Optional[models.Author]]:
-    pass
+    def delete_author(self, id: int) -> None:
+        self._conn.execute(sqlalchemy.text(DELETE_AUTHOR), {"p1": id})
 
+    def get_author(self, id: int) -> Optional[models.Author]:
+        result = self._conn.execute(sqlalchemy.text(GET_AUTHOR), {"p1": id})
+        return models.Author(**dict(result.first()))
 
-def create_author(conn: sqlc.GenericConnection, name: str, bio: Optional[str]) -> sqlc.ReturnType[Optional[models.Author]]:
-    return conn.execute_one_model(models.Author, CREATE_AUTHOR, name, bio)
+    def list_authors(self) -> Iterator[models.Author]:
+        result = self._conn.execute(sqlalchemy.text(LIST_AUTHORS))
+        for row in result:
+            yield models.Author(**dict(row))
 
 
-@overload
-def delete_author(conn: sqlc.Connection, id: int) -> None:
-    pass
+class AsyncQuery:
+    def __init__(self, conn: sqlalchemy.ext.asyncio.AsyncConnection):
+        self._conn = conn
 
+    async def create_author(self, name: str, bio: Optional[str]) -> Optional[models.Author]:
+        result = await self._conn.execute(sqlalchemy.text(CREATE_AUTHOR), {"p1": name, "p2": bio})
+        return models.Author(**dict(result.first()))
 
-@overload
-def delete_author(conn: sqlc.AsyncConnection, id: int) -> Awaitable[None]:
-    pass
-
-
-def delete_author(conn: sqlc.GenericConnection, id: int) -> sqlc.ReturnType[None]:
-    return conn.execute_none(DELETE_AUTHOR, id)
-
-
-@overload
-def get_author(conn: sqlc.Connection, id: int) -> Optional[models.Author]:
-    pass
-
-
-@overload
-def get_author(conn: sqlc.AsyncConnection, id: int) -> Awaitable[Optional[models.Author]]:
-    pass
-
-
-def get_author(conn: sqlc.GenericConnection, id: int) -> sqlc.ReturnType[Optional[models.Author]]:
-    return conn.execute_one_model(models.Author, GET_AUTHOR, id)
-
-
-@overload
-def list_authors(conn: sqlc.Connection) -> Iterator[models.Author]:
-    pass
-
-
-@overload
-def list_authors(conn: sqlc.AsyncConnection) -> AsyncIterator[models.Author]:
-    pass
-
-
-def list_authors(conn: sqlc.GenericConnection) -> sqlc.IteratorReturn[models.Author]:
-    return conn.execute_many_model(models.Author, LIST_AUTHORS)
+    async def delete_author(self, id: int) -> None:
+        await self._conn.execute(sqlalchemy.text(DELETE_AUTHOR), {"p1": id})
 
+    async def get_author(self, id: int) -> Optional[models.Author]:
+        result = await self._conn.execute(sqlalchemy.text(GET_AUTHOR), {"p1": id})
+        return models.Author(**dict(result.first()))
 
+    async def list_authors(self) -> AsyncIterator[models.Author]:
+        result = await self._conn.stream(sqlalchemy.text(LIST_AUTHORS))
+        async for row in result:
+            yield models.Author(**dict(row))

examples/python/src/dbtest/migrations.py

@@ -1,29 +1,26 @@
 import os
 from typing import List
 
-import asyncpg
-import psycopg2.extensions
+import sqlalchemy
+import sqlalchemy.ext.asyncio
 
 
-def apply_migrations(db: psycopg2.extensions.connection, paths: List[str]):
+def apply_migrations(conn: sqlalchemy.engine.Connection, paths: List[str]):
     files = _find_sql_files(paths)
 
     for file in files:
         with open(file, "r") as fd:
             blob = fd.read()
-        cur = db.cursor()
-        cur.execute(blob)
-        cur.close()
-        db.commit()
+        conn.execute(blob)
 
 
-async def apply_migrations_async(db: asyncpg.Connection, paths: List[str]):
+async def apply_migrations_async(conn: sqlalchemy.ext.asyncio.AsyncConnection, paths: List[str]):
     files = _find_sql_files(paths)
 
     for file in files:
         with open(file, "r") as fd:
             blob = fd.read()
-        await db.execute(blob)
+        await conn.execute(sqlalchemy.text(blob))
 
 
 def _find_sql_files(paths: List[str]) -> List[str]:

examples/python/src/tests/conftest.py

@@ -6,6 +6,8 @@
 import psycopg2
 import psycopg2.extensions
 import pytest
+import sqlalchemy
+import sqlalchemy.ext.asyncio
 
 
 @pytest.fixture(scope="session")
@@ -16,7 +18,34 @@ def postgres_uri() -> str:
     pg_password = os.environ.get("PG_PASSWORD", "mysecretpassword")
     pg_db = os.environ.get("PG_DATABASE", "dinotest")
 
-    return f"postgres://{pg_user}:{pg_password}@{pg_host}:{pg_port}/{pg_db}?sslmode=disable"
+    return f"postgresql://{pg_user}:{pg_password}@{pg_host}:{pg_port}/{pg_db}"
+
+
+@pytest.fixture(scope="session")
+def sqlalchemy_connection(postgres_uri) -> sqlalchemy.engine.Connection:
+    schema_name = f"sqltest_{random.randint(0, 1000)}"
+    engine = sqlalchemy.create_engine(postgres_uri)
+    with engine.connect() as conn:
+        conn.execute(f"CREATE SCHEMA {schema_name}")
+        conn.execute(f"SET search_path TO {schema_name}")
+        yield conn
+        conn.execute(f"DROP SCHEMA {schema_name} CASCADE")
+        conn.execute("SET search_path TO public")
+
+
+@pytest.fixture(scope="session")
+async def async_sqlalchemy_connection(postgres_uri) -> sqlalchemy.ext.asyncio.AsyncConnection:
+    postgres_uri = postgres_uri.replace("postgresql", "postgresql+asyncpg")
+    schema_name = f"sqltest_{random.randint(0, 1000)}"
+    engine = sqlalchemy.ext.asyncio.create_async_engine(postgres_uri)
+    async with engine.connect() as conn:
+        await conn.execute(sqlalchemy.text(f"CREATE SCHEMA {schema_name}"))
+        await conn.execute(sqlalchemy.text(f"SET search_path TO {schema_name}"))
+        await conn.commit()
+        yield conn
+        await conn.rollback()
+        await conn.execute(sqlalchemy.text(f"DROP SCHEMA {schema_name} CASCADE"))
+        await conn.execute(sqlalchemy.text("SET search_path TO public"))
 
 
 @pytest.fixture(scope="session")
@@ -29,6 +58,7 @@ def postgres_connection(postgres_uri) -> psycopg2.extensions.connection:
 @pytest.fixture()
 def postgres_db(postgres_connection) -> psycopg2.extensions.connection:
     schema_name = f"sqltest_{random.randint(0, 1000)}"
+    # schema_name = "sqltest_1"
     cur = postgres_connection.cursor()
     cur.execute(f"CREATE SCHEMA {schema_name}")
     cur.execute(f"SET search_path TO {schema_name}")

examples/python/src/tests/test_authors.py

@@ -1,59 +1,56 @@
 import os
 
-import asyncpg
-import psycopg2.extensions
 import pytest
-from sqlc_runtime.psycopg2 import build_psycopg2_connection
-from sqlc_runtime.asyncpg import build_asyncpg_connection
+import sqlalchemy.ext.asyncio
 
 from authors import query
 from dbtest.migrations import apply_migrations, apply_migrations_async
 
 
-def test_authors(postgres_db: psycopg2.extensions.connection):
-    apply_migrations(postgres_db, [os.path.dirname(__file__) + "/../../../authors/postgresql/schema.sql"])
+def test_authors(sqlalchemy_connection: sqlalchemy.engine.Connection):
+    apply_migrations(sqlalchemy_connection, [os.path.dirname(__file__) + "/../../../authors/postgresql/schema.sql"])
 
-    db = build_psycopg2_connection(postgres_db)
+    db = query.Query(sqlalchemy_connection)
 
-    authors = list(query.list_authors(db))
+    authors = list(db.list_authors())
     assert authors == []
 
     author_name = "Brian Kernighan"
     author_bio = "Co-author of The C Programming Language and The Go Programming Language"
-    new_author = query.create_author(db, name=author_name, bio=author_bio)
+    new_author = db.create_author(name=author_name, bio=author_bio)
     assert new_author.id > 0
     assert new_author.name == author_name
     assert new_author.bio == author_bio
 
-    db_author = query.get_author(db, new_author.id)
+    db_author = db.get_author(new_author.id)
     assert db_author == new_author
 
-    author_list = list(query.list_authors(db))
+    author_list = list(db.list_authors())
     assert len(author_list) == 1
     assert author_list[0] == new_author
 
 
 @pytest.mark.asyncio
-async def test_authors_async(async_postgres_db: asyncpg.Connection):
-    await apply_migrations_async(async_postgres_db, [os.path.dirname(__file__) + "/../../../authors/postgresql/schema.sql"])
+async def test_authors_async(async_sqlalchemy_connection: sqlalchemy.ext.asyncio.AsyncConnection):
+    await apply_migrations_async(async_sqlalchemy_connection, [os.path.dirname(__file__) + "/../../../authors/postgresql/schema.sql"])
 
-    db = build_asyncpg_connection(async_postgres_db)
+    db = query.AsyncQuery(async_sqlalchemy_connection)
 
-    async for _ in query.list_authors(db):
+    async for _ in db.list_authors():
         assert False, "No authors should exist"
 
     author_name = "Brian Kernighan"
     author_bio = "Co-author of The C Programming Language and The Go Programming Language"
-    new_author = await query.create_author(db, name=author_name, bio=author_bio)
+    new_author = await db.create_author(name=author_name, bio=author_bio)
     assert new_author.id > 0
     assert new_author.name == author_name
     assert new_author.bio == author_bio
 
-    db_author = await query.get_author(db, new_author.id)
+    db_author = await db.get_author(new_author.id)
     assert db_author == new_author
 
     author_list = []
-    async for author in query.list_authors(db):
+    async for author in db.list_authors():
         author_list.append(author)
     assert len(author_list) == 1
     assert author_list[0] == new_author