(ctx, *args, **kwargs)
| 784 | def decorator(fn): |
| 785 | @wraps(fn) |
| 786 | def wrapper(ctx, *args, **kwargs): |
| 787 | kb_dir = _find_kb_dir(ctx.obj.get("kb_dir_override")) |
| 788 | if kb_dir is None: |
| 789 | return fn(ctx, *args, **kwargs) |
| 790 | if exclusive: |
| 791 | with kb_ingest_lock(kb_dir / ".openkb"): |
| 792 | return fn(ctx, *args, **kwargs) |
| 793 | with kb_read_lock(kb_dir / ".openkb"): |
| 794 | return fn(ctx, *args, **kwargs) |
| 795 | |
| 796 | return wrapper |
| 797 |
nothing calls this directly
no test coverage detected