Skip to content

ccproxy.api.decorators

ccproxy.api.decorators

format_chain 

format_chain(*formats)

Existing simple decorator to attach a format chain to a route handler.

This attaches a format_chain attribute used by validation and helpers.

Source code in ccproxy/api/decorators.py 
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
def format_chain(
    *formats: str,
) -> Callable[[Callable[P, Awaitable[R]]], Callable[P, Awaitable[R]]]:
    """Existing simple decorator to attach a format chain to a route handler.

    This attaches a __format_chain__ attribute used by validation and helpers.
    """

    def decorator(func: Callable[P, Awaitable[R]]) -> Callable[P, Awaitable[R]]:
        func.__format_chain__ = list(formats)  # type: ignore[attr-defined]

        @wraps(func)
        async def wrapper(*args: P.args, **kwargs: P.kwargs) -> R:
            return await func(*args, **kwargs)

        return wrapper

    return decorator

with_format_chain 

with_format_chain(formats, *, endpoint=None)

Decorator to set format chain and optional endpoint metadata on a route.

  • Attaches format_chain to the endpoint for upstream processing/validation
  • Ensures request.state.context exists and sets context.format_chain
  • Optionally sets context.metadata["endpoint"] to the provided upstream endpoint path
Source code in ccproxy/api/decorators.py 
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
def with_format_chain(
    formats: list[str], *, endpoint: str | None = None
) -> Callable[[Callable[P, Awaitable[R]]], Callable[P, Awaitable[R]]]:
    """Decorator to set format chain and optional endpoint metadata on a route.

    - Attaches __format_chain__ to the endpoint for upstream processing/validation
    - Ensures request.state.context exists and sets context.format_chain
    - Optionally sets context.metadata["endpoint"] to the provided upstream endpoint path
    """

    def decorator(func: Callable[P, Awaitable[R]]) -> Callable[P, Awaitable[R]]:
        func.__format_chain__ = list(formats)  # type: ignore[attr-defined]

        @wraps(func)
        async def wrapper(*args: P.args, **kwargs: P.kwargs) -> R:
            # Find Request in args/kwargs
            request: Request | None = None
            for arg in args:
                if isinstance(arg, Request):
                    request = arg
                    break
            if request is None:
                req = kwargs.get("request")
                if isinstance(req, Request):
                    request = req

            if request is not None:
                # Ensure a context exists
                if (
                    not hasattr(request.state, "context")
                    or request.state.context is None
                ):
                    request.state.context = RequestContext(
                        request_id=str(uuid.uuid4()),
                        start_time=time.perf_counter(),
                        logger=_get_logger(__name__),
                    )
                # Set chain and endpoint metadata
                request.state.context.format_chain = list(formats)
                if endpoint:
                    request.state.context.metadata["endpoint"] = endpoint

            return await func(*args, **kwargs)

        return wrapper

    return decorator