0
# AsyncPG
1

2
A high-performance, asyncio-native PostgreSQL database client designed specifically for Python's async/await syntax. AsyncPG implements the PostgreSQL server binary protocol directly, enabling advanced PostgreSQL features like prepared statements, scrollable cursors, and automatic encoding/decoding of composite types while maintaining maximum performance through Cython extensions.
3

4
## Package Information
5

6
- **Package Name**: asyncpg
7
- **Language**: Python
8
- **Installation**: `pip install asyncpg`
9
- **Optional Dependencies**: `pip install asyncpg[gssauth]` for GSSAPI/SSPI authentication
10

11
## Core Imports
12

13
```python
14
import asyncpg
15
```
16

17
Common patterns:
18
```python
19
# Direct connection
20
connection = await asyncpg.connect('postgresql://user:pass@localhost/dbname')
21

22
# Connection pooling
23
pool = asyncpg.create_pool('postgresql://user:pass@localhost/dbname')
24

25
# Exception handling
26
from asyncpg import PostgresError, UniqueViolationError
27
```
28

29
## Basic Usage
30

31
```python
32
import asyncio
33
import asyncpg
34

35
async def main():
36
    # Establish a connection to the database
37
    conn = await asyncpg.connect('postgresql://user:pass@localhost/dbname')
38
    
39
    # Execute a simple query
40
    result = await conn.execute(
41
        "CREATE TABLE users(id serial, name text, email text)"
42
    )
43
    
44
    # Insert data
45
    await conn.execute(
46
        "INSERT INTO users(name, email) VALUES($1, $2)", 
47
        "Alice", "alice@example.com"
48
    )
49
    
50
    # Fetch multiple rows
51
    rows = await conn.fetch("SELECT * FROM users WHERE name = $1", "Alice")
52
    for row in rows:
53
        print(f"User: {row['name']} ({row['email']})")
54
    
55
    # Fetch a single value
56
    user_count = await conn.fetchval("SELECT COUNT(*) FROM users")
57
    print(f"Total users: {user_count}")
58
    
59
    # Clean up
60
    await conn.close()
61

62
# Using connection pool
63
async def pool_example():
64
    pool = asyncpg.create_pool('postgresql://user:pass@localhost/dbname')
65
    
66
    async with pool.acquire() as conn:
67
        result = await conn.fetch("SELECT * FROM users LIMIT 10")
68
        return result
69
    
70
    await pool.close()
71

72
asyncio.run(main())
73
```
74

75
## Architecture
76

77
AsyncPG is built around several core components that provide comprehensive PostgreSQL functionality:
78

79
- **Connection**: Single database session with query execution, transaction management, and type handling
80
- **Pool**: Connection pooling with automatic lifecycle management and load balancing
81
- **Protocol**: Binary protocol implementation optimized for performance with Cython
82
- **Record**: Query result representation with both dict-like and tuple-like access patterns
83
- **Type System**: Automatic encoding/decoding for PostgreSQL data types, arrays, and composite types
84
- **Transaction Management**: Full transaction support including savepoints and isolation levels
85

86
This design enables asyncpg to serve as both a high-level database interface and a foundation for advanced PostgreSQL applications requiring maximum performance and full feature access.
87

88
## Capabilities
89

90
### Connection Management
91

92
Core functionality for establishing and managing database connections, including authentication, SSL/TLS support, and connection lifecycle management.
93

94
```python { .api }
95
async def connect(
96
    dsn: str = None,
97
    *,
98
    host: str = None,
99
    port: int = None,
100
    user: str = None,
101
    password: typing.Union[str, typing.Callable[[], str]] = None,
102
    passfile: str = None,
103
    database: str = None,
104
    loop: asyncio.AbstractEventLoop = None,
105
    timeout: float = 60,
106
    statement_cache_size: int = 100,
107
    max_cached_statement_lifetime: float = 300,
108
    max_cacheable_statement_size: int = 15360,
109
    command_timeout: float = None,
110
    ssl: typing.Union[bool, ssl.SSLContext] = None,
111
    direct_tls: bool = None,
112
    connection_class: type = Connection,
113
    record_class: type = Record,
114
    server_settings: typing.Dict[str, str] = None,
115
    target_session_attrs: str = None,
116
    krbsrvname: str = None,
117
    gsslib: str = None
118
) -> Connection
119
```
120

121
[Connection Management](./connection-management.md)
122

123
### Query Execution
124

125
Comprehensive query execution methods supporting various result formats, parameterized queries, prepared statements, and bulk operations.
126

127
```python { .api }
128
async def execute(self, query: str, *args, timeout: float = None) -> str
129
async def executemany(self, command: str, args, *, timeout: float = None) -> None
130
async def fetch(self, query: str, *args, timeout: float = None, record_class: type = None) -> typing.List[Record]
131
async def fetchval(self, query: str, *args, column: int = 0, timeout: float = None) -> typing.Any
132
async def fetchrow(self, query: str, *args, timeout: float = None, record_class: type = None) -> typing.Optional[Record]
133
async def fetchmany(self, query: str, args, *, timeout: float = None, record_class: type = None) -> typing.List[Record]
134
```
135

136
[Query Execution](./query-execution.md)
137

138
### Prepared Statements
139

140
High-performance reusable query execution with server-side statement caching and optimized parameter binding.
141

142
```python { .api }
143
async def prepare(
144
    self,
145
    query: str,
146
    *,
147
    name: str = None,
148
    timeout: float = None,
149
    record_class: type = None
150
) -> PreparedStatement
151
```
152

153
[Prepared Statements](./prepared-statements.md)
154

155
### Cursor Operations
156

157
Scrollable cursors for efficient traversal of large result sets with configurable prefetch and memory management.
158

159
```python { .api }
160
def cursor(
161
    self,
162
    query: str,
163
    *args,
164
    prefetch: int = None,
165
    timeout: float = None,
166
    record_class: type = None
167
) -> CursorFactory
168
```
169

170
[Cursor Operations](./cursor-operations.md)
171

172
### Connection Pooling
173

174
Advanced connection pooling with configurable pool sizes, connection lifecycle management, and transparent connection acquisition/release.
175

176
```python { .api }
177
def create_pool(
178
    dsn: str = None,
179
    *,
180
    min_size: int = 10,
181
    max_size: int = 10,
182
    max_queries: int = 50000,
183
    max_inactive_connection_lifetime: float = 300.0,
184
    connect: typing.Callable = None,
185
    setup: typing.Callable = None,
186
    init: typing.Callable = None,
187
    reset: typing.Callable = None,
188
    loop: asyncio.AbstractEventLoop = None,
189
    connection_class: type = Connection,
190
    record_class: type = Record,
191
    **connect_kwargs
192
) -> Pool
193
```
194

195
[Connection Pooling](./connection-pooling.md)
196

197
### Transaction Management
198

199
Full transaction support including transaction contexts, savepoints, isolation levels, and read-only transactions.
200

201
```python { .api }
202
def transaction(
203
    self, 
204
    *, 
205
    isolation: str = None, 
206
    readonly: bool = False, 
207
    deferrable: bool = False
208
) -> Transaction: ...
209
```
210

211
[Transaction Management](./transaction-management.md)
212

213
### COPY Operations
214

215
High-performance bulk data import/export using PostgreSQL's COPY protocol with support for various formats and streaming.
216

217
```python { .api }
218
async def copy_from_table(
219
    self,
220
    table_name: str,
221
    *,
222
    output,
223
    columns: typing.List[str] = None,
224
    schema_name: str = None,
225
    timeout: float = None,
226
    format: str = None,
227
    oids: bool = None,
228
    delimiter: str = None,
229
    null: str = None,
230
    header: bool = None,
231
    quote: str = None,
232
    escape: str = None,
233
    force_quote: typing.Union[bool, typing.List[str]] = None,
234
    encoding: str = None
235
) -> str
236

237
async def copy_from_query(
238
    self,
239
    query: str,
240
    *args,
241
    output,
242
    timeout: float = None,
243
    format: str = None,
244
    oids: bool = None,
245
    delimiter: str = None,
246
    null: str = None,
247
    header: bool = None,
248
    quote: str = None,
249
    escape: str = None,
250
    force_quote: typing.Union[bool, typing.List[str]] = None,
251
    encoding: str = None
252
) -> str
253

254
async def copy_to_table(
255
    self,
256
    table_name: str,
257
    *,
258
    source,
259
    columns: typing.List[str] = None,
260
    schema_name: str = None,
261
    timeout: float = None,
262
    format: str = None,
263
    oids: bool = None,
264
    freeze: bool = None,
265
    delimiter: str = None,
266
    null: str = None,
267
    header: bool = None,
268
    quote: str = None,
269
    escape: str = None,
270
    force_quote: typing.Union[bool, typing.List[str]] = None,
271
    force_not_null: typing.List[str] = None,
272
    force_null: typing.List[str] = None,
273
    encoding: str = None,
274
    where: str = None
275
) -> str
276

277
async def copy_records_to_table(
278
    self,
279
    table_name: str,
280
    *,
281
    records,
282
    columns: typing.List[str] = None,
283
    schema_name: str = None,
284
    timeout: float = None,
285
    where: str = None
286
) -> str
287
```
288

289
[COPY Operations](./copy-operations.md)
290

291
### Type System and Codecs
292

293
Comprehensive type system supporting all PostgreSQL data types, custom type registration, and automatic encoding/decoding.
294

295
```python { .api }
296
async def set_type_codec(
297
    self,
298
    typename: str,
299
    *,
300
    schema: str = 'public',
301
    encoder: typing.Callable,
302
    decoder: typing.Callable,
303
    format: str = 'text'
304
) -> None
305

306
async def reset_type_codec(
307
    self,
308
    typename: str,
309
    *,
310
    schema: str = 'public'
311
) -> None
312

313
async def set_builtin_type_codec(
314
    self,
315
    typename: str,
316
    *,
317
    schema: str = 'public',
318
    codec_name: str,
319
    format: str = None
320
) -> None
321

322
async def reload_schema_state(self) -> None
323
```
324

325
[Type System](./type-system.md)
326

327
### Exception Handling
328

329
Complete exception hierarchy mapping PostgreSQL error codes to Python exceptions with detailed error information.
330

331
```python { .api }
332
class PostgresError(Exception): ...
333
class FatalPostgresError(PostgresError): ...
334
class InterfaceError(Exception): ...
335
class InterfaceWarning(UserWarning): ...
336
class DataError(InterfaceError, ValueError): ...
337
class IntegrityConstraintViolationError(PostgresError): ...
338
class UniqueViolationError(IntegrityConstraintViolationError): ...
339
class ForeignKeyViolationError(IntegrityConstraintViolationError): ...
340
class NotNullViolationError(IntegrityConstraintViolationError): ...
341
class CheckViolationError(IntegrityConstraintViolationError): ...
342
```
343

344
[Exception Handling](./exception-handling.md)
345

346
### Listeners and Notifications
347

348
Support for PostgreSQL's LISTEN/NOTIFY functionality and server log message handling.
349

350
```python { .api }
351
async def add_listener(self, channel: str, callback: typing.Callable) -> None
352
async def remove_listener(self, channel: str, callback: typing.Callable) -> None
353
def add_log_listener(self, callback: typing.Callable) -> None
354
def remove_log_listener(self, callback: typing.Callable) -> None
355
def add_termination_listener(self, callback: typing.Callable) -> None
356
def remove_termination_listener(self, callback: typing.Callable) -> None
357
def add_query_logger(self, callback: typing.Callable) -> None
358
def remove_query_logger(self, callback: typing.Callable) -> None
359
```
360

361
[Listeners and Notifications](./listeners-notifications.md)
362

363
## Types
364

365
### Core Connection Types
366

367
```python { .api }
368
class Connection:
369
    """A representation of a database session."""
370
    
371
    def is_closed(self) -> bool
372
    def get_server_pid(self) -> int
373
    def get_server_version(self) -> ServerVersion
374
    def get_settings(self) -> ConnectionSettings
375
    def is_in_transaction(self) -> bool
376
    async def close(self, *, timeout: float = None) -> None
377
    async def reset(self, *, timeout: float = None) -> None
378
    def terminate(self) -> None
379
    def get_reset_query(self) -> str
380

381
class Pool:
382
    """A connection pool."""
383
    
384
    def get_size(self) -> int
385
    def get_min_size(self) -> int
386
    def get_max_size(self) -> int
387
    def get_idle_size(self) -> int
388
    def is_closing(self) -> bool
389
    async def acquire(self, *, timeout: float = None) -> PoolAcquireContext
390
    async def release(self, connection: Connection, *, timeout: float = None) -> None
391
    async def close(self) -> None
392
    def terminate(self) -> None
393
    async def expire_connections(self) -> None
394
    def set_connect_args(self, dsn: str = None, **connect_kwargs) -> None
395

396
class PreparedStatement:
397
    """A prepared statement."""
398
    
399
    def get_name(self) -> str
400
    def get_query(self) -> str
401
    def get_statusmsg(self) -> str
402
    def get_parameters(self) -> typing.Tuple[Type, ...]
403
    def get_attributes(self) -> typing.Tuple[Attribute, ...]
404

405
class CursorFactory:
406
    """Factory for creating cursors."""
407
    
408
    def __aiter__(self) -> CursorIterator
409
    def __await__(self) -> Cursor
410

411
class Transaction:
412
    """A transaction context manager."""
413
    
414
    async def start(self) -> None
415
    async def commit(self) -> None
416
    async def rollback(self) -> None
417
    async def __aenter__(self) -> Transaction
418
    async def __aexit__(self, extype, ex, tb) -> None
419
```
420

421
### Query Result Types
422

423
```python { .api }
424
class Record:
425
    """Query result record with dict-like and tuple-like access."""
426
    
427
    def get(self, key: str, default=None): ...
428
    def keys(self) -> Iterator[str]: ...
429
    def values(self) -> Iterator: ...
430
    def items(self) -> Iterator[tuple[str, typing.Any]]: ...
431
    def __getitem__(self, key: typing.Union[str, int, slice]): ...
432
    def __len__(self) -> int: ...
433
```
434

435
### PostgreSQL Data Types
436

437
```python { .api }
438
class Type:
439
    """Database data type information."""
440
    oid: int
441
    name: str
442
    kind: str
443
    schema: str
444

445
class Attribute:
446
    """Database attribute information."""
447
    name: str
448
    type: Type
449

450
class ServerVersion:
451
    """PostgreSQL server version tuple."""
452
    major: int
453
    minor: int
454
    micro: int
455
    releaselevel: str
456
    serial: int
457

458
class Range:
459
    """PostgreSQL range type representation."""
460
    lower: typing.Any
461
    upper: typing.Any
462
    lower_inc: bool
463
    upper_inc: bool
464
    lower_inf: bool
465
    upper_inf: bool
466
    isempty: bool
467
    
468
    def issubset(self, other: Range) -> bool
469
    def issuperset(self, other: Range) -> bool
470

471
class BitString:
472
    """PostgreSQL bit string type."""
473
    value: str
474

475
class Point:
476
    """PostgreSQL point geometric type."""
477
    x: float
478
    y: float
479

480
class Box:
481
    """PostgreSQL box geometric type."""
482
    high: Point
483
    low: Point
484

485
class Path:
486
    """PostgreSQL path geometric type."""
487
    is_closed: bool
488
    points: typing.List[Point]
489

490
class Polygon:
491
    """PostgreSQL polygon geometric type."""
492
    points: typing.List[Point]
493

494
class Line:
495
    """PostgreSQL line geometric type."""
496
    a: float
497
    b: float
498
    c: float
499

500
class LineSegment:
501
    """PostgreSQL line segment geometric type."""
502
    p1: Point
503
    p2: Point
504

505
class Circle:
506
    """PostgreSQL circle geometric type."""
507
    center: Point
508
    radius: float
509

510
class ConnectionSettings:
511
    """Connection configuration and settings."""
512
    ...
513
```