(name, **kwargs)
| 17 | |
| 18 | @staticmethod |
| 19 | def get(name, **kwargs): |
| 20 | |
| 21 | if name in ["mysql", "oceanbase"]: |
| 22 | from modelcache_mm.manager.scalar_data.sql_storage import SQLStorage |
| 23 | config = kwargs.get("config") |
| 24 | import_sql_client(name) |
| 25 | cache_base = SQLStorage(db_type=name, config=config) |
| 26 | elif name == 'sqlite': |
| 27 | from modelcache_mm.manager.scalar_data.sql_storage_sqlite import SQLStorage |
| 28 | sql_url = kwargs.get("sql_url", SQL_URL[name]) |
| 29 | cache_base = SQLStorage(db_type=name, url=sql_url) |
| 30 | else: |
| 31 | raise NotFoundError("cache store", name) |
| 32 | return cache_base |
no test coverage detected