0
# Database Connections
1

2
Client fixture factory for creating database connections with automatic cleanup, transaction isolation, and database management between tests. Provides psycopg connections with comprehensive lifecycle management.
3

4
## Capabilities
5

6
### Client Fixture Factory
7

8
Creates database connection fixtures that automatically manage database lifecycle and cleanup.
9

10
```python { .api }
11
def postgresql(
12
    process_fixture_name: str,
13
    dbname: Optional[str] = None,
14
    isolation_level: Optional[psycopg.IsolationLevel] = None,
15
) -> Callable[[FixtureRequest], Iterator[Connection]]:
16
    """
17
    Create a postgresql client fixture factory.
18

19
    Parameters:
20
    - process_fixture_name: Name of the process or noproc fixture to depend on
21
    - dbname: Override database name (default: uses process fixture's dbname)
22
    - isolation_level: Transaction isolation level (default: server default)
23

24
    Returns:
25
    Function that creates psycopg.Connection fixture with function scope
26
    """
27
```
28

29
### Database Janitor
30

31
Manages database lifecycle including creation, population, and cleanup.
32

33
```python { .api }
34
class DatabaseJanitor:
35
    """
36
    Database lifecycle manager for test isolation.
37
    
38
    Handles database creation, initialization, and cleanup to ensure
39
    test isolation and repeatability.
40
    """
41
    
42
    def __init__(
43
        self,
44
        *,
45
        user: str,
46
        host: str,
47
        port: Union[str, int],
48
        version: Union[str, float, Version],
49
        dbname: str,
50
        template_dbname: str,
51
        password: Optional[str] = None,
52
        isolation_level: Optional[psycopg.IsolationLevel] = None,
53
        connection_timeout: int = 60,
54
    ): ...
55
    
56
    def init(self) -> None: ...
57
    def drop(self) -> None: ...
58
    def load(self, load_list: List[Union[Callable, str, Path]]) -> None: ...
59
    def cursor(self) -> Iterator[Cursor]: ...
60
    def is_template(self) -> bool: ...
61
    
62
    def __enter__(self) -> "DatabaseJanitor": ...
63
    def __exit__(
64
        self,
65
        exc_type: Optional[Type[BaseException]],
66
        exc_val: Optional[BaseException], 
67
        exc_tb: Optional[TracebackType],
68
    ) -> None: ...
69
```
70

71
## Usage Examples
72

73
### Basic Client Connection
74

75
```python
76
from pytest_postgresql import factories
77

78
# Create process and client fixtures
79
postgresql_proc = factories.postgresql_proc()
80
postgresql = factories.postgresql('postgresql_proc')
81

82
def test_database_connection(postgresql):
83
    """Test basic database connection."""
84
    cur = postgresql.cursor()
85
    cur.execute("SELECT version();")
86
    result = cur.fetchone()
87
    assert result is not None
88
    cur.close()
89
```
90

91
### Custom Database Name
92

93
```python
94
from pytest_postgresql import factories
95

96
postgresql_proc = factories.postgresql_proc(dbname='main_db')
97
postgresql_custom = factories.postgresql('postgresql_proc', dbname='custom_test_db')
98

99
def test_custom_database(postgresql_custom, postgresql_proc):
100
    """Test connection to custom database."""
101
    cur = postgresql_custom.cursor()
102
    cur.execute("SELECT current_database();")
103
    db_name = cur.fetchone()[0]
104
    assert db_name == 'custom_test_db'
105
    cur.close()
106
```
107

108
### Transaction Isolation
109

110
```python
111
import psycopg
112
from pytest_postgresql import factories
113

114
postgresql_proc = factories.postgresql_proc()
115
postgresql_serializable = factories.postgresql(
116
    'postgresql_proc',
117
    isolation_level=psycopg.IsolationLevel.SERIALIZABLE
118
)
119

120
def test_transaction_isolation(postgresql_serializable):
121
    """Test transaction isolation level."""
122
    cur = postgresql_serializable.cursor()
123
    cur.execute("SHOW transaction_isolation;")
124
    isolation = cur.fetchone()[0]
125
    assert isolation == 'serializable'
126
    cur.close()
127
```
128

129
### Multiple Database Connections
130

131
```python
132
from pytest_postgresql import factories
133

134
postgresql_proc = factories.postgresql_proc()
135
postgresql_db1 = factories.postgresql('postgresql_proc', dbname='db1')
136
postgresql_db2 = factories.postgresql('postgresql_proc', dbname='db2')
137

138
def test_multiple_databases(postgresql_db1, postgresql_db2):
139
    """Test connections to multiple databases."""
140
    # Create table in first database
141
    cur1 = postgresql_db1.cursor()
142
    cur1.execute("CREATE TABLE test1 (id INT);")
143
    postgresql_db1.commit()
144
    cur1.close()
145
    
146
    # Create different table in second database
147
    cur2 = postgresql_db2.cursor()
148
    cur2.execute("CREATE TABLE test2 (id INT);")
149
    postgresql_db2.commit()
150
    cur2.close()
151
    
152
    # Verify isolation between databases
153
    cur1 = postgresql_db1.cursor()
154
    cur1.execute("SELECT tablename FROM pg_tables WHERE schemaname='public';")
155
    tables1 = [row[0] for row in cur1.fetchall()]
156
    assert 'test1' in tables1
157
    assert 'test2' not in tables1
158
    cur1.close()
159
```
160

161
### Database Operations
162

163
```python
164
def test_database_operations(postgresql):
165
    """Test various database operations."""
166
    cur = postgresql.cursor()
167
    
168
    # Create table
169
    cur.execute("""
170
        CREATE TABLE users (
171
            id SERIAL PRIMARY KEY,
172
            name VARCHAR(100) NOT NULL,
173
            email VARCHAR(100) UNIQUE,
174
            created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
175
        );
176
    """)
177
    
178
    # Insert data
179
    cur.execute(
180
        "INSERT INTO users (name, email) VALUES (%s, %s) RETURNING id;",
181
        ('John Doe', 'john@example.com')
182
    )
183
    user_id = cur.fetchone()[0]
184
    
185
    # Query data
186
    cur.execute("SELECT name, email FROM users WHERE id = %s;", (user_id,))
187
    user = cur.fetchone()
188
    assert user[0] == 'John Doe'
189
    assert user[1] == 'john@example.com'
190
    
191
    # Update data
192
    cur.execute(
193
        "UPDATE users SET name = %s WHERE id = %s;",
194
        ('Jane Doe', user_id)
195
    )
196
    
197
    # Delete data
198
    cur.execute("DELETE FROM users WHERE id = %s;", (user_id,))
199
    
200
    postgresql.commit()
201
    cur.close()
202
```
203

204
### Context Manager Usage
205

206
```python
207
def test_context_manager(postgresql):
208
    """Test using connection as context manager."""
209
    with postgresql:
210
        with postgresql.cursor() as cur:
211
            cur.execute("CREATE TABLE temp_table (data TEXT);")
212
            cur.execute("INSERT INTO temp_table (data) VALUES ('test');")
213
            
214
            cur.execute("SELECT data FROM temp_table;")
215
            result = cur.fetchone()
216
            assert result[0] == 'test'
217
    # Connection automatically committed/rolled back
218
```
219

220
### Connection Properties
221

222
```python
223
def test_connection_properties(postgresql):
224
    """Test connection properties and information."""
225
    # Connection status
226
    assert postgresql.status == psycopg.pq.ConnStatus.CONNECTION_OK
227
    
228
    # Server information
229
    info = postgresql.info
230
    assert info.server_version > 0
231
    assert info.host is not None
232
    assert info.port > 0
233
    
234
    # Database information
235
    cur = postgresql.cursor()
236
    cur.execute("SELECT current_database(), current_user, version();")
237
    db_name, user, version = cur.fetchone()
238
    assert db_name is not None
239
    assert user is not None
240
    assert 'PostgreSQL' in version
241
    cur.close()
242
```
243

244
### Error Handling
245

246
```python
247
import psycopg
248
from psycopg import errors
249

250
def test_error_handling(postgresql):
251
    """Test database error handling."""
252
    cur = postgresql.cursor()
253
    
254
    try:
255
        # This will fail - table doesn't exist
256
        cur.execute("SELECT * FROM nonexistent_table;")
257
    except errors.UndefinedTable as e:
258
        assert 'nonexistent_table' in str(e)
259
    
260
    try:
261
        # This will fail - syntax error
262
        cur.execute("INVALID SQL SYNTAX;")
263
    except errors.SyntaxError as e:
264
        assert 'syntax error' in str(e).lower()
265
    
266
    # Connection should still be usable after errors
267
    cur.execute("SELECT 1;")
268
    result = cur.fetchone()
269
    assert result[0] == 1
270
    
271
    cur.close()
272
```
273

274
### Batch Operations
275

276
```python
277
def test_batch_operations(postgresql):
278
    """Test batch database operations."""
279
    cur = postgresql.cursor()
280
    
281
    # Create table
282
    cur.execute("""
283
        CREATE TABLE products (
284
            id SERIAL PRIMARY KEY,
285
            name VARCHAR(100),
286
            price DECIMAL(10,2)
287
        );
288
    """)
289
    
290
    # Batch insert
291
    products = [
292
        ('Product A', 19.99),
293
        ('Product B', 29.99),
294
        ('Product C', 39.99)
295
    ]
296
    
297
    cur.executemany(
298
        "INSERT INTO products (name, price) VALUES (%s, %s);",
299
        products
300
    )
301
    
302
    # Verify batch insert
303
    cur.execute("SELECT COUNT(*) FROM products;")
304
    count = cur.fetchone()[0]
305
    assert count == 3
306
    
307
    postgresql.commit()
308
    cur.close()
309
```
310

311
## Connection Lifecycle
312

313
### Automatic Cleanup
314

315
The client fixture automatically handles cleanup between tests:
316

317
1. **After each test**: Closes all open connections and drops the test database
318
2. **Template database**: Maintains a template database for fast test database creation
319
3. **Isolation**: Each test gets a fresh database instance
320

321
### Manual Database Management
322

323
```python
324
from pytest_postgresql.janitor import DatabaseJanitor
325

326
def test_manual_database_management(postgresql_proc):
327
    """Test manual database management with DatabaseJanitor."""
328
    janitor = DatabaseJanitor(
329
        user=postgresql_proc.user,
330
        host=postgresql_proc.host,
331
        port=postgresql_proc.port,
332
        version=postgresql_proc.version,
333
        dbname='manual_test_db',
334
        template_dbname=postgresql_proc.template_dbname,
335
        password=postgresql_proc.password
336
    )
337
    
338
    with janitor:
339
        # Database is created and ready
340
        with janitor.cursor() as cur:
341
            cur.execute("CREATE TABLE manual_test (id INT);")
342
            cur.execute("INSERT INTO manual_test (id) VALUES (1);")
343
            
344
            cur.execute("SELECT id FROM manual_test;")
345
            result = cur.fetchone()
346
            assert result[0] == 1
347
    # Database is automatically dropped
348
```