Skip to content

Commit d4a83bc

Browse files
ZxMYSaoen
authored andcommitted
[AIRFLOW-XXXX] Expose SQLAlchemy's connect_args and make it configurable (apache#6478)
In many use cases users need to configure SQLAlchemy's connect_args (e.g. pass ssl.check_hostname=False to PyMySQL), and Airflow should expose this option and make it configurable.
1 parent f68c186 commit d4a83bc

File tree

4 files changed

+125
-4
lines changed

4 files changed

+125
-4
lines changed

CONTRIBUTING.rst

+4-2
Original file line numberDiff line numberDiff line change
@@ -285,11 +285,13 @@ To fix a pylint issue, do the following:
285285
1. Remove module/modules from the
286286
`scripts/ci/pylint_todo.txt <scripts/ci/pylint_todo.txt>`__.
287287

288-
2. Run `scripts/ci/ci_pylint.sh <scripts/ci/ci_pylint.sh>`__.
288+
2. Run `scripts/ci/ci_pylint_main.sh <scripts/ci/ci_pylint_main.sh>`__ and
289+
`scripts/ci/ci_pylint_tests.sh <scripts/ci/ci_pylint_tests.sh>`__.
289290

290291
3. Fix all the issues reported by pylint.
291292

292-
4. Re-run `scripts/ci/ci_pylint.sh <scripts/ci/ci_pylint.sh>`__.
293+
4. Re-run `scripts/ci/ci_pylint_main.sh <scripts/ci/ci_pylint_main.sh>`__ and
294+
`scripts/ci/ci_pylint_tests.sh <scripts/ci/ci_pylint_tests.sh>`__.
293295

294296
5. If you see "success", submit a PR following
295297
`Pull Request guidelines <#pull-request-guidelines>`__.

airflow/config_templates/default_airflow.cfg

+6-1
Original file line numberDiff line numberDiff line change
@@ -131,6 +131,11 @@ sql_alchemy_pool_pre_ping = True
131131
# SqlAlchemy supports databases with the concept of multiple schemas.
132132
sql_alchemy_schema =
133133

134+
# Import path for connect args in SqlAlchemy. Default to an empty dict.
135+
# This is useful when you want to configure db engine args that SqlAlchemy won't parse in connection string.
136+
# See https://docs.sqlalchemy.org/en/13/core/engines.html#sqlalchemy.create_engine.params.connect_args
137+
# sql_alchemy_connect_args =
138+
134139
# The amount of parallelism as a setting to the executor. This defines
135140
# the max number of task instances that should run simultaneously
136141
# on this airflow installation
@@ -378,7 +383,7 @@ smtp_mail_from = airflow@example.com
378383

379384
[sentry]
380385
# Sentry (https://docs.sentry.io) integration
381-
sentry_dsn =
386+
sentry_dsn =
382387

383388

384389
[celery]

airflow/settings.py

+9-1
Original file line numberDiff line numberDiff line change
@@ -33,6 +33,7 @@
3333
import airflow
3434
from airflow.configuration import AIRFLOW_HOME, WEBSERVER_CONFIG, conf # NOQA F401
3535
from airflow.logging_config import configure_logging
36+
from airflow.utils.module_loading import import_string
3637
from airflow.utils.sqlalchemy import setup_event_handlers
3738

3839
log = logging.getLogger(__name__)
@@ -185,7 +186,14 @@ def configure_orm(disable_connection_pool=False):
185186
# For Python2 we get back a newstr and need a str
186187
engine_args['encoding'] = engine_args['encoding'].__str__()
187188

188-
engine = create_engine(SQL_ALCHEMY_CONN, **engine_args)
189+
if conf.has_option('core', 'sql_alchemy_connect_args'):
190+
connect_args = import_string(
191+
conf.get('core', 'sql_alchemy_connect_args')
192+
)
193+
else:
194+
connect_args = {}
195+
196+
engine = create_engine(SQL_ALCHEMY_CONN, connect_args=connect_args, **engine_args)
189197
setup_event_handlers(engine)
190198

191199
Session = scoped_session(

tests/test_sqlalchemy_config.py

+106
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,106 @@
1+
# -*- coding: utf-8 -*-
2+
#
3+
# Licensed to the Apache Software Foundation (ASF) under one
4+
# or more contributor license agreements. See the NOTICE file
5+
# distributed with this work for additional information
6+
# regarding copyright ownership. The ASF licenses this file
7+
# to you under the Apache License, Version 2.0 (the
8+
# "License"); you may not use this file except in compliance
9+
# with the License. You may obtain a copy of the License at
10+
#
11+
# http://www.apache.org/licenses/LICENSE-2.0
12+
#
13+
# Unless required by applicable law or agreed to in writing,
14+
# software distributed under the License is distributed on an
15+
# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
16+
# KIND, either express or implied. See the License for the
17+
# specific language governing permissions and limitations
18+
# under the License.
19+
20+
import unittest
21+
22+
from sqlalchemy.pool import NullPool
23+
24+
from airflow import settings
25+
from tests.compat import patch
26+
from tests.test_utils.config import conf_vars
27+
28+
SQL_ALCHEMY_CONNECT_ARGS = {
29+
'test': 43503,
30+
'dict': {
31+
'is': 1,
32+
'supported': 'too'
33+
}
34+
}
35+
36+
37+
class TestSqlAlchemySettings(unittest.TestCase):
38+
def setUp(self):
39+
self.old_engine = settings.engine
40+
self.old_session = settings.Session
41+
self.old_conn = settings.SQL_ALCHEMY_CONN
42+
settings.SQL_ALCHEMY_CONN = "mysql+foobar://user:pass@host/dbname?inline=param&another=param"
43+
44+
def tearDown(self):
45+
settings.engine = self.old_engine
46+
settings.Session = self.old_session
47+
settings.SQL_ALCHEMY_CONN = self.old_conn
48+
49+
@patch('airflow.settings.setup_event_handlers')
50+
@patch('airflow.settings.scoped_session')
51+
@patch('airflow.settings.sessionmaker')
52+
@patch('airflow.settings.create_engine')
53+
def test_configure_orm_with_default_values(self,
54+
mock_create_engine,
55+
mock_sessionmaker,
56+
mock_scoped_session,
57+
mock_setup_event_handlers):
58+
settings.configure_orm()
59+
mock_create_engine.assert_called_once_with(
60+
settings.SQL_ALCHEMY_CONN,
61+
connect_args={},
62+
encoding='utf-8',
63+
max_overflow=10,
64+
pool_pre_ping=True,
65+
pool_recycle=1800,
66+
pool_size=5
67+
)
68+
69+
@patch('airflow.settings.setup_event_handlers')
70+
@patch('airflow.settings.scoped_session')
71+
@patch('airflow.settings.sessionmaker')
72+
@patch('airflow.settings.create_engine')
73+
def test_sql_alchemy_connect_args(self,
74+
mock_create_engine,
75+
mock_sessionmaker,
76+
mock_scoped_session,
77+
mock_setup_event_handlers):
78+
config = {
79+
('core', 'sql_alchemy_connect_args'): 'tests.test_sqlalchemy_config.SQL_ALCHEMY_CONNECT_ARGS',
80+
('core', 'sql_alchemy_pool_enabled'): 'False'
81+
}
82+
with conf_vars(config):
83+
settings.configure_orm()
84+
mock_create_engine.assert_called_once_with(
85+
settings.SQL_ALCHEMY_CONN,
86+
connect_args=SQL_ALCHEMY_CONNECT_ARGS,
87+
poolclass=NullPool,
88+
encoding='utf-8'
89+
)
90+
91+
@patch('airflow.settings.setup_event_handlers')
92+
@patch('airflow.settings.scoped_session')
93+
@patch('airflow.settings.sessionmaker')
94+
@patch('airflow.settings.create_engine')
95+
def test_sql_alchemy_invalid_connect_args(self,
96+
mock_create_engine,
97+
mock_sessionmaker,
98+
mock_scoped_session,
99+
mock_setup_event_handlers):
100+
config = {
101+
('core', 'sql_alchemy_connect_args'): 'does.not.exist',
102+
('core', 'sql_alchemy_pool_enabled'): 'False'
103+
}
104+
with self.assertRaises(ImportError):
105+
with conf_vars(config):
106+
settings.configure_orm()

0 commit comments

Comments
 (0)