(ds: CoreDatasource, timeout: int = 0)
| 137 | |
| 138 | # use sqlalchemy |
| 139 | def get_engine(ds: CoreDatasource, timeout: int = 0) -> Engine: |
| 140 | conf = DatasourceConf(**json.loads(aes_decrypt(ds.configuration))) if not equals_ignore_case(ds.type, |
| 141 | "excel") else get_engine_config() |
| 142 | if conf.timeout is None: |
| 143 | conf.timeout = timeout |
| 144 | if timeout > 0: |
| 145 | conf.timeout = timeout |
| 146 | |
| 147 | if equals_ignore_case(ds.type, "pg"): |
| 148 | if conf.dbSchema is not None and conf.dbSchema != "": |
| 149 | engine = create_engine(get_uri(ds), |
| 150 | connect_args={"options": f"-c search_path={urllib.parse.quote(conf.dbSchema)}", |
| 151 | "connect_timeout": conf.timeout}, poolclass=NullPool) |
| 152 | else: |
| 153 | engine = create_engine(get_uri(ds), connect_args={"connect_timeout": conf.timeout}, poolclass=NullPool) |
| 154 | elif equals_ignore_case(ds.type, 'sqlServer'): |
| 155 | engine = create_engine('mssql+pymssql://', creator=lambda: get_origin_connect(ds.type, conf), |
| 156 | poolclass=NullPool) |
| 157 | elif equals_ignore_case(ds.type, 'oracle'): |
| 158 | engine = create_engine(get_uri(ds), poolclass=NullPool) |
| 159 | elif equals_ignore_case(ds.type, 'mysql'): # mysql |
| 160 | ssl_mode = {"require": True} if conf.ssl else None |
| 161 | engine = create_engine(get_uri(ds), connect_args={"connect_timeout": conf.timeout, "ssl": ssl_mode}, |
| 162 | poolclass=NullPool) |
| 163 | else: # ck |
| 164 | engine = create_engine(get_uri(ds), connect_args={"connect_timeout": conf.timeout}, poolclass=NullPool) |
| 165 | return engine |
| 166 | |
| 167 | |
| 168 | def get_session(ds: CoreDatasource | AssistantOutDsSchema): |
no test coverage detected