2
2
3
3
import psycopg
4
4
import psycopg_pool
5
+ from psycopg .rows import namedtuple_row
5
6
from sqlalchemy .engine .interfaces import Dialect
6
7
from sqlalchemy .dialects .postgresql .psycopg import PGDialect_psycopg
7
8
from sqlalchemy .sql import ClauseElement
@@ -58,12 +59,11 @@ def connection(self) -> "PsycopgConnection":
58
59
class PsycopgConnection (ConnectionBackend ):
59
60
_database : PsycopgBackend
60
61
_dialect : Dialect
61
- _connection : typing .Optional [psycopg .AsyncConnection ]
62
+ _connection : typing .Optional [psycopg .AsyncConnection ] = None
62
63
63
64
def __init__ (self , database : PsycopgBackend , dialect : Dialect ) -> None :
64
65
self ._database = database
65
66
self ._dialect = dialect
66
- self ._connection = None
67
67
68
68
async def acquire (self ) -> None :
69
69
if self ._connection is not None :
@@ -74,6 +74,7 @@ async def acquire(self) -> None:
74
74
75
75
# TODO: Add configurable timeouts
76
76
self ._connection = await self ._database ._pool .getconn ()
77
+ await self ._connection .set_autocommit (True )
77
78
78
79
async def release (self ) -> None :
79
80
if self ._connection is None :
@@ -88,7 +89,7 @@ async def fetch_all(self, query: ClauseElement) -> typing.List[RecordInterface]:
88
89
89
90
query_str , args , result_columns = self ._compile (query )
90
91
91
- async with self ._connection .cursor () as cursor :
92
+ async with self ._connection .cursor (row_factory = namedtuple_row ) as cursor :
92
93
await cursor .execute (query_str , args )
93
94
rows = await cursor .fetchall ()
94
95
@@ -101,7 +102,7 @@ async def fetch_one(self, query: ClauseElement) -> typing.Optional[RecordInterfa
101
102
102
103
query_str , args , result_columns = self ._compile (query )
103
104
104
- async with self ._connection .cursor () as cursor :
105
+ async with self ._connection .cursor (row_factory = namedtuple_row ) as cursor :
105
106
await cursor .execute (query_str , args )
106
107
row = await cursor .fetchone ()
107
108
@@ -127,7 +128,7 @@ async def execute(self, query: ClauseElement) -> typing.Any:
127
128
128
129
query_str , args , _ = self ._compile (query )
129
130
130
- async with self ._connection .cursor () as cursor :
131
+ async with self ._connection .cursor (row_factory = namedtuple_row ) as cursor :
131
132
await cursor .execute (query_str , args )
132
133
133
134
async def execute_many (self , queries : typing .List [ClauseElement ]) -> None :
@@ -144,7 +145,7 @@ async def iterate(
144
145
query_str , args , result_columns = self ._compile (query )
145
146
column_maps = create_column_maps (result_columns )
146
147
147
- async with self ._connection .cursor () as cursor :
148
+ async with self ._connection .cursor (row_factory = namedtuple_row ) as cursor :
148
149
await cursor .execute (query_str , args )
149
150
150
151
while True :
0 commit comments