-
-
Notifications
You must be signed in to change notification settings - Fork 13
Expand file tree
/
Copy pathconfig.py
More file actions
342 lines (317 loc) · 16.3 KB
/
config.py
File metadata and controls
342 lines (317 loc) · 16.3 KB
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
# SPDX-FileCopyrightText: 2020-present The Firebird Projects <www.firebirdsql.org>
#
# SPDX-License-Identifier: MIT
#
# PROGRAM/MODULE: firebird-driver
# FILE: firebird/driver/config.py
# DESCRIPTION: Driver configuration
# CREATED: 1.6.2020
#
# The contents of this file are subject to the MIT License
#
# Permission is hereby granted, free of charge, to any person obtaining a copy
# of this software and associated documentation files (the "Software"), to deal
# in the Software without restriction, including without limitation the rights
# to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
# copies of the Software, and to permit persons to whom the Software is
# furnished to do so, subject to the following conditions:
#
# The above copyright notice and this permission notice shall be included in all
# copies or substantial portions of the Software.
#
# THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
# IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
# FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
# AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
# LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
# OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
# SOFTWARE.
#
# Copyright (c) 2020 Firebird Project (www.firebirdsql.org)
# All Rights Reserved.
#
# Contributor(s): Pavel Císař (original code)
# ______________________________________
"""firebird-driver - Driver configuration
This module defines the configuration system for the firebird-driver.
It uses an INI-style format managed via the `DriverConfig` class, which
allows defining settings for the driver itself, default server/database
parameters, and named configurations for specific servers and databases.
Configuration can be loaded from files, strings, or dictionaries, and
supports environment variable interpolation. The primary interaction point
is usually the global `driver_config` instance.
"""
from __future__ import annotations
import os
from collections.abc import Iterable
from firebird.base.config import (
BoolOption,
Config,
ConfigListOption,
ConfigParser,
EnumOption,
EnvExtendedInterpolation,
IntOption,
ListOption,
StrOption,
)
from .types import DecfloatRound, DecfloatTraps, NetProtocol
class ServerConfig(Config):
"""Represents configuration options specific to a named Firebird server entry.
"""
def __init__(self, name: str, *, optional: bool=False, description: str | None=None):
super().__init__(name, optional=optional, description=description)
#: Server host machine specification
self.host: StrOption = \
StrOption('host', "Server host machine specification")
#: Port used by Firebird server
self.port: StrOption = \
StrOption('port', "Port used by Firebird server")
#: Defaul user name, default is envar ISC_USER or None if not specified
self.user: StrOption = \
StrOption('user', "Defaul user name", default=os.environ.get('ISC_USER', None))
#: Default user password, default is envar ISC_PASSWORD or None if not specified
self.password: StrOption = \
StrOption('password', "Default user password",
default=os.environ.get('ISC_PASSWORD', None))
#: Configuration override
self.config: StrOption = \
StrOption('config', "Configuration override")
#: List of authentication plugins override
self.auth_plugin_list: StrOption = \
StrOption('auth_plugin_list', "List of authentication plugins override")
#: Use trusted authentication, default: False
self.trusted_auth: BoolOption = \
BoolOption('trusted_auth', "Use trusted authentication", default=False)
#: Encoding used for text data exchange with server
self.encoding: StrOption = \
StrOption('encoding', "Encoding used for text data exchange with server",
default='ascii')
#: Handler used for encoding errors. See `codecs error handlers <codecs>` for details.
self.encoding_errors: StrOption = \
StrOption('encoding_errors', "Handler used for encoding errors", default='strict')
class DatabaseConfig(Config):
"""Represents configuration options specific to a named Firebird database entry, including connection and creation parameters.
"""
def __init__(self, name: str, *, optional: bool=False, description: str | None=None):
super().__init__(name, optional=optional, description=description)
#: Name of server where database is located
self.server: StrOption = \
StrOption('server', "Name of server where database is located")
#: Database connection string
self.dsn: StrOption = \
StrOption('dsn', "Database connection string")
#: Database file specification or alias
self.database: StrOption = \
StrOption('database', "Database file specification or alias")
#: Specifies whether the database parameter (filename) is encoded in UTF-8 when passed to the server.
self.utf8filename: BoolOption = \
BoolOption('utf8filename', "Specifies whether the database parameter (filename) is encoded in UTF-8 when passed to the server.")
#: Network protocol to use for the connection. Value must be a member of the .NetProtocol enum.
self.protocol: EnumOption = \
EnumOption('protocol', NetProtocol, "Network protocol to use for the connection. Value must be a member of the .NetProtocol enum.")
#: Defaul user name, default is envar ISC_USER or None if not specified
self.user: StrOption = \
StrOption('user', "Defaul user name", default=os.environ.get('ISC_USER', None))
#: Default user password, default is envar ISC_PASSWORD or None if not specified
self.password: StrOption = \
StrOption('password', "Default user password",
default=os.environ.get('ISC_PASSWORD', None))
#: Use trusted authentication, default: False
self.trusted_auth: BoolOption = \
BoolOption('trusted_auth', "Use trusted authentication", default=False)
#: User role
self.role: StrOption = \
StrOption('role', "User role")
#: Character set for database connection
self.charset: StrOption = \
StrOption('charset', "Character set for database connection")
#: SQL Dialect for database connection, default: 3
self.sql_dialect: IntOption = \
IntOption('sql_dialect', "SQL Dialect for database connection", default=3)
#: Connection timeout
self.timeout: IntOption = \
IntOption('timeout', "Connection timeout")
#: Do not use linger for database connection
self.no_linger: BoolOption = \
BoolOption('no_linger', "Do not use linger for database connection")
#: Page cache size override for database connection
self.cache_size: IntOption = \
IntOption('cache_size', "Page cache size override for database connection")
#: Dummy packet interval for this database connection
self.dummy_packet_interval: IntOption = \
IntOption('dummy_packet_interval',
"Dummy packet interval")
#: Configuration override
self.config: StrOption = \
StrOption('config', "Configuration override")
#: List of authentication plugins override
self.auth_plugin_list: StrOption = \
StrOption('auth_plugin_list', "List of authentication plugins override")
#: Session time zone [Firebird 4]
self.session_time_zone: StrOption = \
StrOption('session_time_zone', "Session time zone")
#: Set BIND [Firebird 4]
self.set_bind: StrOption = \
StrOption('set_bind', "Set BIND - sets up columns coercion rules in session")
#: Set DECFLOAT ROUND [Firebird 4], value is `.DecfloatRound`
self.decfloat_round: EnumOption = \
EnumOption('decfloat_round', DecfloatRound, "DECFLOAT round mode")
#: Specifies which DECFLOAT exceptional conditions should cause a trap (raise an error) [Firebird 4]. Accepts a list of .DecfloatTraps enum members.
self.decfloat_traps: ListOption = \
ListOption('decfloat_traps', DecfloatTraps,
"""Specifies which DECFLOAT exceptional conditions should cause a trap
(raise an error) [Firebird 4]. Accepts a list of .DecfloatTraps enum members.""")
#: Number of parallel workers
self.parallel_workers = \
IntOption('parallel_workers', "Number of parallel workers")
# --- Options specific to Database Creation ---
#: Database create option. Page size to be used.
self.page_size: IntOption = \
IntOption('page_size', "Page size to be used for created database.")
#: Database create option. Write mode (True = sync/False = async).
self.forced_writes: BoolOption = \
BoolOption('forced_writes', "Write mode for created database (True = sync, False = async)")
#: Database create option. Character set for the database.
self.db_charset: StrOption = \
StrOption('db_charset', "Character set for created database")
#: Database create option. SQL dialect for the database.
self.db_sql_dialect: IntOption = \
IntOption('db_sql_dialect', "SQL dialect for created database")
#: Database create option. Page cache size override for database.
self.db_cache_size: IntOption = \
IntOption('db_cache_size', "Page cache size override for created database")
#: Database create option. Sweep interval for the database.
self.sweep_interval: IntOption = \
IntOption('sweep_interval', "Sweep interval for created database")
#: Database create option. Data page space usage (True = reserve space, False = Use all space).
self.reserve_space: BoolOption = \
BoolOption('reserve_space',
"Data page space usage for created database (True = reserve space, False = Use all space)")
class DriverConfig(Config):
"""Main configuration object for the Firebird driver. Holds global settings, default
configurations, and lists of registered servers and databases.
Settings loaded from specific server/database sections override those in the `db_defaults`
and `server_defaults` sections. When reading multiple files, options from later files
override those from earlier ones.
"""
def __init__(self, name: str):
super().__init__(name)
#: Path to Firebird client library
self.fb_client_library: StrOption = \
StrOption('fb_client_library', "Path to Firebird client library")
#: BLOB size threshold. Bigger BLOB will be returned as stream BLOBs.
self.stream_blob_threshold: IntOption = \
IntOption('stream_blob_threshold',
"BLOB size threshold. Bigger BLOB will be returned as stream BLOBs.",
default=65536)
#: Default database configuration ('firebird.db.defaults')
self.db_defaults: DatabaseConfig = DatabaseConfig('firebird.db.defaults',
optional=True,
description="Default database configuration.")
#: Default server configuration ('firebird.server.defaults')
self.server_defaults: ServerConfig = ServerConfig('firebird.server.defaults',
optional=True,
description="Default server configuration.")
#: Registered servers
self.servers: ConfigListOption = \
ConfigListOption('servers', ServerConfig, "Registered servers")
#: Registered databases
self.databases: ConfigListOption = \
ConfigListOption('databases', DatabaseConfig, "Registered databases")
def read(self, filenames: str | Iterable, encoding: str | None=None):
"""Read configuration from a filename or an iterable of filenames.
Files that cannot be opened are silently ignored; this is
designed so that you can specify an iterable of potential
configuration file locations (e.g. current directory, user's
home directory, systemwide directory), and all existing
configuration files in the iterable will be read. A single
filename may also be given.
Return list of successfully read files.
"""
parser = ConfigParser(interpolation=EnvExtendedInterpolation())
read_ok = parser.read(filenames, encoding)
if read_ok:
self.load_config(parser)
return read_ok
def read_file(self, f):
"""Read configuration from a file-like object.
The `f` argument must be iterable, returning one line at a time.
"""
parser = ConfigParser(interpolation=EnvExtendedInterpolation())
parser.read_file(f)
self.load_config(parser)
def read_string(self, string: str) -> None:
"""Read configuration from a given string.
"""
parser = ConfigParser(interpolation=EnvExtendedInterpolation())
parser.read_string(string)
self.load_config(parser)
def read_dict(self, dictionary: dict[str, str]) -> None:
"""Read configuration from a dictionary.
Keys are section names, values are dictionaries with keys and values
that should be present in the section. If the used dictionary type
preserves order, sections and their keys will be added in order.
All types held in the dictionary are converted to strings during
reading, including section names, option names and keys.
"""
parser = ConfigParser(interpolation=EnvExtendedInterpolation())
parser.read_dict(dictionary)
self.load_config(parser)
def get_server(self, name: str) -> ServerConfig:
"""Returns server configuration.
"""
for srv in self.servers.value:
if srv.name == name:
return srv
return None
def get_database(self, name: str) -> DatabaseConfig:
"""Returns database configuration.
"""
for db in self.databases.value:
if db.name == name:
return db
return None
def register_server(self, name: str, config: str | None=None) -> ServerConfig:
"""Register server.
Arguments:
name: Server name.
config: Optional server configuration string in ConfigParser format in [name] section.
Returns:
ServerConfig: For newly registered server
Raises:
ValueError: If server is already registered.
"""
if self.get_server(name) is not None:
raise ValueError(f"Server '{name}' already registered.")
srv_config = ServerConfig(name)
self.servers.value.append(srv_config)
if config:
parser = ConfigParser(interpolation=EnvExtendedInterpolation())
parser.read_string(config)
srv_config.load_config(parser, name)
return srv_config
def register_database(self, name: str, config: str | None=None) -> DatabaseConfig:
"""Register database.
Arguments:
name: Database name.
config: Optional database configuration string in ConfigParser format in [name] section.
Returns:
DatabaseConfig: For newly registered database
Raises:
ValueError: If database is already registered.
"""
if self.get_database(name) is not None:
raise ValueError(f"Database '{name}' already registered.")
db_config = DatabaseConfig(name)
self.databases.value.append(db_config)
if config:
parser = ConfigParser(interpolation=EnvExtendedInterpolation())
parser.read_string(config)
db_config.load_config(parser, name)
return db_config
# Configuration
#: Global driver configuration instance.
#: Load settings from files/strings/dicts into this object before connecting,
#: or modify its attributes directly for programmatic configuration.
driver_config: DriverConfig = DriverConfig('firebird.driver')