Coverage for src/qdrant_loader/connectors/factory.py: 92%

1import inspect 

2from typing import Any 

3 

4from .registry import CONNECTOR_REGISTRY 

5 

6 

7def get_connector_instance(config: Any, checkpoint_cursor: str | None = None) -> Any: 

8 """Create and return a connector instance based on the config's source and deployment type. 

9 

10 Accepts an optional `checkpoint_cursor` which will be passed to connector 

11 implementations that support resumption (e.g., Jira connectors). If the 

12 connector class does not accept the extra parameter, the function falls 

13 back to constructing it with the single `config` argument for 

14 backward-compatibility. 

15 """ 

16 key = (config.source_type, getattr(config, "deployment_type", None)) 

17 

18 connector_class = CONNECTOR_REGISTRY.get(key) 

19 

20 if not connector_class: 

21 raise ValueError(f"Unsupported connector type: {key}") 

22 

23 # Try to construct with checkpoint_cursor kwarg; fall back to single-arg. 

24 init_params = inspect.signature(connector_class).parameters 

25 if "checkpoint_cursor" in init_params: 

26 return connector_class(config, checkpoint_cursor=checkpoint_cursor) 

27 return connector_class(config)