12
12
from airbyte_cdk .destinations import Destination
13
13
from airbyte_cdk .models import AirbyteConnectionStatus , AirbyteMessage , ConfiguredAirbyteCatalog , DestinationSyncMode , Status , Type
14
14
from firebolt .client import DEFAULT_API_URL
15
- from firebolt .client .auth import UsernamePassword , ClientCredentials , Auth
15
+ from firebolt .client .auth import Auth , ClientCredentials , UsernamePassword
16
16
from firebolt .db import Connection , connect
17
17
18
18
from .writer import create_firebolt_wirter
19
19
20
20
logger = getLogger ("airbyte" )
21
21
22
22
23
- def _determine_auth (key : str , secret :str ) -> Auth :
23
+ def _determine_auth (key : str , secret : str ) -> Auth :
24
24
"""
25
25
Determine between new auth based on key and secret or legacy email based auth.
26
26
"""
27
- if '@' in key :
27
+ if "@" in key :
28
28
# email auth can only be used with UsernamePassword
29
29
return UsernamePassword (key , secret )
30
30
else :
31
31
return ClientCredentials (key , secret )
32
32
33
+
33
34
def parse_config (config : json , logger : Optional [AirbyteLogger ] = None ) -> Dict [str , Any ]:
34
35
"""
35
36
Convert dict of config values to firebolt.db.Connection arguments
@@ -38,8 +39,7 @@ def parse_config(config: json, logger: Optional[AirbyteLogger] = None) -> Dict[s
38
39
:return: dictionary of firebolt.db.Connection-compatible kwargs
39
40
"""
40
41
# We should use client_id/client_secret, this code supports username/password for legacy users
41
- auth = _determine_auth (config .get ("client_id" , config .get ("username" )),
42
- config .get ("client_secret" , config .get ("password" )))
42
+ auth = _determine_auth (config .get ("client_id" , config .get ("username" )), config .get ("client_secret" , config .get ("password" )))
43
43
connection_args = {
44
44
"database" : config ["database" ],
45
45
"auth" : auth ,
0 commit comments