1
1
import json
2
2
from subprocess import CalledProcessError , run
3
- from typing import Any , Dict , Optional , cast
3
+ from typing import Optional
4
4
5
5
from .exception import RSConnectException
6
6
from .log import logger
@@ -13,7 +13,7 @@ def snow(*args: str):
13
13
14
14
def ensure_snow_installed () -> None :
15
15
try :
16
- import snowflake .cli
16
+ import snowflake .cli # noqa
17
17
18
18
logger .debug ("snowflake-cli is installed." )
19
19
@@ -33,7 +33,7 @@ def list_connections():
33
33
res = snow ("connection" , "list" , "--format" , "json" )
34
34
connection_list = json .loads (res .stdout )
35
35
return connection_list
36
- except :
36
+ except CalledProcessError :
37
37
raise RSConnectException ("Could not list snowflake connections." )
38
38
39
39
@@ -64,7 +64,7 @@ def generate_jwt(name: Optional[str] = None):
64
64
res = snow ("connection" , "generate-jwt" , "--connection" , connection_name , "--format" , "json" )
65
65
try :
66
66
output = json .loads (res .stdout )
67
- except json .JSONDecodeError as e :
67
+ except json .JSONDecodeError :
68
68
raise RSConnectException (f"Failed to parse JSON from snow-cli: { res .stdout } " )
69
69
jwt = output .get ("message" )
70
70
if jwt is None :
0 commit comments