(ds: CoreDatasource | AssistantOutDsSchema)
| 166 | |
| 167 | |
| 168 | def get_session(ds: CoreDatasource | AssistantOutDsSchema): |
| 169 | # engine = get_engine(ds) if isinstance(ds, CoreDatasource) else get_ds_engine(ds) |
| 170 | if isinstance(ds, AssistantOutDsSchema): |
| 171 | out_conf = get_out_ds_conf(ds, 30) |
| 172 | ds.configuration = out_conf |
| 173 | |
| 174 | engine = get_engine(ds) |
| 175 | session_maker = sessionmaker(bind=engine) |
| 176 | session = session_maker() |
| 177 | return session |
| 178 | |
| 179 | |
| 180 | def check_connection(trans: Optional[Trans], ds: CoreDatasource | AssistantOutDsSchema, is_raise: bool = False): |
no test coverage detected