Skip to content

Stream Cursors #35

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Draft
vgvoleg wants to merge 1 commit into
base: main
Choose a base branch
from
Draft

Stream Cursors #35

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
22 changes: 22 additions & 0 deletions README.md
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -37,6 +37,17 @@ with connection.cursor() as cursor:
rows = cursor.fetchall()
```

For large result sets you can use a server-side cursor that streams result
sets incrementally instead of buffering the whole response in memory:
Comment on lines +40 to +41
Copy link

Copilot AI Apr 24, 2026

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

The docs describe how to enable stream_results, but they don’t mention the key behavioral constraint enforced by the code: during interactive transactions, an active streaming cursor can block other queries and commit/rollback until the stream is fully consumed or explicitly closed (raising ProgrammingError otherwise). Consider adding a short note here (and in the async section) explaining that callers must consume all rows or call cursor.close() before issuing other operations on the same connection/transaction.

Copilot uses AI. Check for mistakes.

```python
with connection.cursor(stream_results=True) as cursor:
cursor.execute("SELECT id, val FROM table")

for row in iter(cursor.fetchone, None):
...
```

Usage of async connection:

```python
Expand All @@ -47,3 +58,14 @@ async with async_connection.cursor() as cursor:
rows = await cursor.fetchmany(size=5)
rows = await cursor.fetchall()
```

Async streaming cursors are enabled with the same flag:

```python
async with async_connection.cursor(stream_results=True) as cursor:
await cursor.execute("SELECT id, val FROM table")

row = await cursor.fetchone()
rows = await cursor.fetchmany(size=5)
rows = await cursor.fetchall()
```
6 changes: 3 additions & 3 deletions poetry.lock
View file
Open in desktop

Some generated files are not rendered by default. Learn more about how customized files appear on GitHub.

26 changes: 10 additions & 16 deletions tests/test_connections.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -65,7 +65,6 @@ def _test_commit_rollback_after_begin(
maybe_await(connection.begin())
maybe_await(connection.rollback())


def _test_connection(self, connection: dbapi.Connection) -> None:
maybe_await(connection.commit())
maybe_await(connection.rollback())
Expand Down Expand Up @@ -100,9 +99,12 @@ def _test_cursor_raw_query(self, connection: dbapi.Connection) -> None:
with suppress(dbapi.DatabaseError):
maybe_await(cur.execute_scheme("DROP TABLE test"))

maybe_await(cur.execute_scheme(
"CREATE TABLE test(id Int64 NOT NULL, text Utf8, PRIMARY KEY (id))"
))
maybe_await(
cur.execute_scheme(
"CREATE TABLE test("
"id Int64 NOT NULL, text Utf8, PRIMARY KEY (id))"
)
)

maybe_await(
cur.execute(
Expand Down Expand Up @@ -421,9 +423,7 @@ def test_commit_rollback_after_begin(
isolation_level: str,
connection: dbapi.Connection,
) -> None:
self._test_commit_rollback_after_begin(
connection, isolation_level
)
self._test_commit_rollback_after_begin(connection, isolation_level)

def test_connection(self, connection: dbapi.Connection) -> None:
self._test_connection(connection)
Expand All @@ -442,14 +442,10 @@ def test_errors_with_interactive_tx(
) -> None:
self._test_error_with_interactive_tx(connection)

def test_get_view_names(
self, connection: dbapi.Connection
) -> None:
def test_get_view_names(self, connection: dbapi.Connection) -> None:
self._test_get_view_names(connection)

def test_get_table_names(
self, connection: dbapi.Connection
) -> None:
def test_get_table_names(self, connection: dbapi.Connection) -> None:
self._test_get_table_names(connection)


Expand Down Expand Up @@ -527,9 +523,7 @@ async def test_commit_rollback_after_begin(
connection: dbapi.AsyncConnection,
) -> None:
await greenlet_spawn(
self._test_commit_rollback_after_begin,
connection,
isolation_level
self._test_commit_rollback_after_begin, connection, isolation_level
)

@pytest.mark.asyncio
Expand Down
Loading
Loading