Coverage for agentos/checkpoint/factory.py: 50%

1""" 

2Checkpointer 工厂 — 统一创建不同后端的 Checkpointer。 

3""" 

4 

5from __future__ import annotations 

6 

7from typing import Any 

8 

9from agentos.checkpoint.base import CheckpointBackend 

10from agentos.checkpoint.sqlite import SQLiteCheckpointer 

11from agentos.checkpoint.postgres import PostgresCheckpointer 

12 

13__all__ = ["create_checkpointer"] 

14 

15 

16def create_checkpointer( 

17 backend: str = "sqlite", 

18 **kwargs: Any, 

19) -> CheckpointBackend: 

20 """ 

21 创建 Checkpointer 实例。 

22 

23 Args: 

24 backend: "sqlite" | "postgres" 

25 **kwargs: 后端特定参数 

26 - sqlite: db_path (默认 "checkpoints.db") 

27 - postgres: dsn (默认 "postgresql://localhost:5432/agentos") 

28 

29 Returns: 

30 CheckpointBackend 实例。 

31 

32 Raises: 

33 ValueError: 未知后端。 

34 """ 

35 backend = backend.lower().strip() 

36 

37 if backend == "sqlite": 

38 db_path = kwargs.pop("db_path", "checkpoints.db") 

39 return SQLiteCheckpointer(db_path=db_path, **kwargs) 

40 

41 if backend == "postgres": 

42 return PostgresCheckpointer(**kwargs) 

43 

44 raise ValueError( 

45 f"Unknown checkpoint backend: '{backend}'. Available: sqlite, postgres" 

46 )