0
# Connection and Configuration
1
2
Connection management, URL building utilities, and configuration options for optimizing Snowflake database connectivity within SQLAlchemy applications.
3
4
## Capabilities
5
6
### Connection URL Building
7
8
Utilities for constructing proper Snowflake connection strings and URLs.
9
10
```python { .api }
11
from snowflake.sqlalchemy import URL
12
from snowflake.sqlalchemy.util import parse_url_boolean, parse_url_integer
13
14
def URL(**db_parameters):
15
"""
16
Compose SQLAlchemy connection string for Snowflake.
17
18
Args:
19
**db_parameters: Database connection parameters
20
21
Returns:
22
str: SQLAlchemy connection URL
23
"""
24
25
def parse_url_boolean(value: str) -> bool:
26
"""
27
Parse URL boolean values from connection strings.
28
29
Args:
30
value: String value to parse
31
32
Returns:
33
bool: Parsed boolean value
34
"""
35
36
def parse_url_integer(value: str) -> int:
37
"""
38
Parse URL integer values from connection strings.
39
40
Args:
41
value: String value to parse
42
43
Returns:
44
int: Parsed integer value
45
"""
46
```
47
48
### Connection Management
49
50
Internal utilities for managing connections and application settings.
51
52
```python { .api }
53
from snowflake.sqlalchemy.util import (
54
_update_connection_application_name,
55
_set_connection_interpolate_empty_sequences
56
)
57
58
def _update_connection_application_name(**conn_kwargs):
59
"""
60
Update connection application name for telemetry.
61
62
Args:
63
**conn_kwargs: Connection keyword arguments
64
65
Returns:
66
dict: Updated connection arguments
67
"""
68
69
def _set_connection_interpolate_empty_sequences(connection, flag: bool):
70
"""
71
Set connection interpolation flag for empty sequences.
72
73
Args:
74
connection: Database connection
75
flag: Whether to interpolate empty sequences
76
"""
77
```
78
79
### Exception Classes
80
81
Comprehensive exception hierarchy for connection and configuration errors.
82
83
```python { .api }
84
from sqlalchemy.exc import ArgumentError
85
from snowflake.sqlalchemy.exc import (
86
NoPrimaryKeyError,
87
UnsupportedPrimaryKeysAndForeignKeysError,
88
RequiredParametersNotProvidedError,
89
UnexpectedTableOptionKeyError,
90
OptionKeyNotProvidedError,
91
UnexpectedOptionParameterTypeError,
92
CustomOptionsAreOnlySupportedOnSnowflakeTables,
93
UnexpectedOptionTypeError,
94
InvalidTableParameterTypeError,
95
MultipleErrors,
96
StructuredTypeNotSupportedInTableColumnsError
97
)
98
99
class NoPrimaryKeyError(ArgumentError):
100
"""Raised when primary key is required but missing."""
101
102
def __init__(self, target: str):
103
"""
104
Initialize with target table name.
105
106
Args:
107
target: Target table name
108
"""
109
110
class UnsupportedPrimaryKeysAndForeignKeysError(ArgumentError):
111
"""Raised when primary keys or foreign keys are not supported."""
112
113
def __init__(self, target: str):
114
"""
115
Initialize with target table name.
116
117
Args:
118
target: Target table name
119
"""
120
121
class RequiredParametersNotProvidedError(ArgumentError):
122
"""Raised when required parameters are not provided."""
123
124
def __init__(self, target: str, parameters: List[str]):
125
"""
126
Initialize with target and missing parameters.
127
128
Args:
129
target: Target object name
130
parameters: List of missing parameter names
131
"""
132
133
class UnexpectedTableOptionKeyError(ArgumentError):
134
"""Raised when an unexpected table option key is used."""
135
136
def __init__(self, expected: str, actual: str):
137
"""
138
Initialize with expected and actual option keys.
139
140
Args:
141
expected: Expected option key
142
actual: Actual option key received
143
"""
144
145
class OptionKeyNotProvidedError(ArgumentError):
146
"""Raised when an option key is not provided."""
147
148
def __init__(self, target: str):
149
"""
150
Initialize with target name.
151
152
Args:
153
target: Target object name
154
"""
155
156
class UnexpectedOptionParameterTypeError(ArgumentError):
157
"""Raised when an option parameter has wrong type."""
158
159
def __init__(self, parameter_name: str, target: str, types: List[str]):
160
"""
161
Initialize with parameter details.
162
163
Args:
164
parameter_name: Parameter name
165
target: Target object name
166
types: List of expected types
167
"""
168
169
class CustomOptionsAreOnlySupportedOnSnowflakeTables(ArgumentError):
170
"""Raised when custom options are used on non-Snowflake tables."""
171
172
class UnexpectedOptionTypeError(ArgumentError):
173
"""Raised when an unexpected option type is encountered."""
174
175
def __init__(self, options: List[str]):
176
"""
177
Initialize with unsupported options.
178
179
Args:
180
options: List of unsupported option names
181
"""
182
183
class InvalidTableParameterTypeError(ArgumentError):
184
"""Raised when table parameter has invalid type."""
185
186
def __init__(self, name: str, input_type: str, expected_types: List[str]):
187
"""
188
Initialize with parameter type details.
189
190
Args:
191
name: Parameter name
192
input_type: Actual input type
193
expected_types: List of expected types
194
"""
195
196
class MultipleErrors(ArgumentError):
197
"""Container for multiple errors."""
198
199
def __init__(self, errors: List[Exception]):
200
"""
201
Initialize with multiple errors.
202
203
Args:
204
errors: List of exceptions
205
"""
206
207
class StructuredTypeNotSupportedInTableColumnsError(ArgumentError):
208
"""Raised when structured types are used inappropriately."""
209
210
def __init__(self, table_type: str, table_name: str, column_name: str):
211
"""
212
Initialize with table and column details.
213
214
Args:
215
table_type: Type of table
216
table_name: Name of table
217
column_name: Name of column with structured type
218
"""
219
```
220
221
### Version Information
222
223
Package version access and metadata.
224
225
```python { .api }
226
from snowflake.sqlalchemy.version import VERSION
227
import snowflake.sqlalchemy
228
229
VERSION: str = "1.7.6"
230
__version__ = snowflake.sqlalchemy.__version__
231
```
232
233
## Usage Examples
234
235
### Basic Connection
236
237
```python
238
from sqlalchemy import create_engine
239
from snowflake.sqlalchemy import URL
240
241
# Direct connection string
242
engine = create_engine(
243
'snowflake://user:password@account_identifier/database/schema'
244
)
245
246
# Using URL builder
247
connection_params = {
248
'user': 'myuser',
249
'password': 'mypassword',
250
'account': 'myaccount',
251
'database': 'mydatabase',
252
'schema': 'myschema',
253
'warehouse': 'mywarehouse',
254
'role': 'myrole'
255
}
256
257
engine = create_engine(URL(**connection_params))
258
```
259
260
### Advanced Connection Configuration
261
262
```python
263
from sqlalchemy import create_engine
264
265
# Connection with additional parameters
266
engine = create_engine(
267
'snowflake://user:password@account.region.cloud/db/schema',
268
connect_args={
269
'warehouse': 'COMPUTE_WH',
270
'role': 'ANALYST_ROLE',
271
'client_session_keep_alive': True,
272
'autocommit': False,
273
'numpy': True
274
}
275
)
276
```
277
278
### Connection with Authentication Options
279
280
```python
281
# Key-pair authentication
282
engine = create_engine(
283
'snowflake://user@account/database/schema',
284
connect_args={
285
'private_key_path': '/path/to/private_key.p8',
286
'private_key_passphrase': 'key_passphrase'
287
}
288
)
289
290
# SSO authentication
291
engine = create_engine(
292
'snowflake://user@account/database/schema',
293
connect_args={
294
'authenticator': 'externalbrowser'
295
}
296
)
297
298
# OAuth authentication
299
engine = create_engine(
300
'snowflake://user@account/database/schema',
301
connect_args={
302
'authenticator': 'oauth',
303
'token': 'oauth_access_token'
304
}
305
)
306
```
307
308
### URL Parameter Parsing
309
310
```python
311
from snowflake.sqlalchemy.util import parse_url_boolean, parse_url_integer
312
313
# Parse boolean parameters from connection string
314
autocommit = parse_url_boolean('true') # True
315
numpy_support = parse_url_boolean('false') # False
316
317
# Parse integer parameters
318
login_timeout = parse_url_integer('30') # 30
319
network_timeout = parse_url_integer('300') # 300
320
```
321
322
### Connection Pooling Configuration
323
324
```python
325
from sqlalchemy import create_engine
326
from sqlalchemy.pool import QueuePool
327
328
# Engine with connection pooling
329
engine = create_engine(
330
'snowflake://user:password@account/db/schema',
331
poolclass=QueuePool,
332
pool_size=5,
333
max_overflow=10,
334
pool_pre_ping=True,
335
pool_recycle=3600
336
)
337
```
338
339
### Error Handling
340
341
```python
342
from sqlalchemy import create_engine
343
from snowflake.sqlalchemy.exc import (
344
RequiredParametersNotProvidedError,
345
InvalidTableParameterTypeError
346
)
347
348
try:
349
# Connection attempt
350
engine = create_engine('snowflake://incomplete_url')
351
352
# Table creation with validation
353
from snowflake.sqlalchemy import HybridTable
354
355
# This would raise NoPrimaryKeyError for HybridTable
356
table = HybridTable('test', metadata, Column('id', Integer))
357
358
except RequiredParametersNotProvidedError as e:
359
print(f"Missing required parameters: {e}")
360
except InvalidTableParameterTypeError as e:
361
print(f"Invalid table parameter: {e}")
362
```
363
364
### Application Name and Telemetry
365
366
```python
367
from snowflake.sqlalchemy.util import _update_connection_application_name
368
369
# Update connection for custom application
370
conn_params = {
371
'user': 'myuser',
372
'password': 'mypassword',
373
'account': 'myaccount'
374
}
375
376
# Add application name for telemetry
377
updated_params = _update_connection_application_name(
378
application='MyDataApp',
379
version='1.0.0',
380
**conn_params
381
)
382
383
engine = create_engine(URL(**updated_params))
384
```
385
386
### Connection Testing and Validation
387
388
```python
389
from sqlalchemy import create_engine, text
390
391
def test_snowflake_connection(connection_string):
392
"""Test Snowflake connection and return version."""
393
try:
394
engine = create_engine(connection_string)
395
with engine.connect() as conn:
396
result = conn.execute(text('SELECT CURRENT_VERSION()')).fetchone()
397
print(f"Connected to Snowflake version: {result[0]}")
398
return True
399
except Exception as e:
400
print(f"Connection failed: {e}")
401
return False
402
finally:
403
engine.dispose()
404
405
# Test connection
406
success = test_snowflake_connection(
407
'snowflake://user:password@account/database/schema'
408
)
409
```
410
411
### Version Information
412
413
```python
414
import snowflake.sqlalchemy
415
from snowflake.sqlalchemy.version import VERSION
416
417
# Get package version
418
print(f"Snowflake SQLAlchemy version: {snowflake.sqlalchemy.__version__}")
419
print(f"Version constant: {VERSION}")
420
421
# Version-dependent feature checks
422
if tuple(map(int, snowflake.sqlalchemy.__version__.split('.'))) >= (1, 7, 0):
423
print("Dynamic tables supported")
424
```