http

The http proxy lib.

`SUPPORTED_HTTP_VERSIONS = ('1.0', '1.1')` `module-attribute` ¶

The http versions that we supported now. It depends on httpx.

`BaseHttpProxy` ¶

Bases: BaseProxyModel

Http proxy base class.

Attributes:

Name	Type	Description
`client`		The `httpx.AsyncClient` to send http requests.
`follow_redirects`		Whether follow redirects of target server.

Source code in src/fastapi_proxy_lib/core/http.py

class BaseHttpProxy(BaseProxyModel):
    """Http proxy base class.

    Attributes:
        client: The `httpx.AsyncClient` to send http requests.
        follow_redirects: Whether follow redirects of target server.
    """

    @override
    async def send_request_to_target(  # pyright: ignore [reportIncompatibleMethodOverride]
        self, *, request: StarletteRequest, target_url: httpx.URL
    ) -> StarletteResponse:
        """Change request headers and send request to target url.

        - The http version of request must be in [`SUPPORTED_HTTP_VERSIONS`][fastapi_proxy_lib.core.http.SUPPORTED_HTTP_VERSIONS].

        Args:
            request: the original client request.
            target_url: target url that request will be sent to.

        Returns:
            The response from target url.
        """
        client = self.client
        follow_redirects = self.follow_redirects

        check_result = check_http_version(request.scope, SUPPORTED_HTTP_VERSIONS)
        if check_result is not None:
            return check_result

        # 将请求头中的host字段改为目标url的host
        # 同时强制移除"keep-alive"字段和添加"keep-alive"值到"connection"字段中保持连接
        require_close, proxy_header = _change_client_header(
            headers=request.headers, target_url=target_url
        )

        # 有些方法不应该包含主体
        request_content = (
            None if request.method in _NON_REQUEST_BODY_METHODS else request.stream()
        )

        # FIX: https://github.com/WSH032/fastapi-proxy-lib/security/advisories/GHSA-7vwr-g6pm-9hc8
        # time cost: 396 ns ± 3.39 ns
        # 由于这不是原子性的操作，所以不保证一定阻止cookie泄漏
        # 一定能保证修复的方法是通过`_tool.change_necessary_client_header_for_httpx`强制指定优先级最高的cookie头
        client.cookies.clear()

        # NOTE: 不要在这里catch `client.build_request` 和 `client.send` 的异常，因为通常来说
        # - 反向代理的异常需要报 5xx 错误
        # - 而正向代理的异常需要报 4xx 错误
        proxy_request = client.build_request(
            method=request.method,
            url=target_url,
            params=request.query_params,
            headers=proxy_header,
            content=request_content,  # FIXME: 一个已知问题是，流式响应头包含'transfer-encoding': 'chunked'，但有些服务器会400拒绝这个头
            # cookies=request.cookies,  # NOTE: headers中已有的cookie优先级高，所以这里不需要
        )

        # DEBUG: 用于调试的记录
        logging.debug(
            "HTTP: client:%s ; url:%s ; head:%s",
            request.client,
            proxy_request.url,
            proxy_request.headers,
        )

        proxy_response = await client.send(
            proxy_request,
            stream=True,
            follow_redirects=follow_redirects,
        )

        tasks = BackgroundTasks()
        tasks.add_task(proxy_response.aclose)  # 添加后端任务，使其在响应完后关闭

        # 依据先前客户端的请求，决定是否要添加"connection": "close"头到响应头中以关闭连接
        # https://www.uvicorn.org/server-behavior/#http-headers
        # 如果响应头包含"connection": "close"，uvicorn会自动关闭连接
        proxy_response_headers = _change_server_header(
            headers=proxy_response.headers, require_close=require_close
        )
        return StreamingResponse(
            content=proxy_response.aiter_raw(),
            status_code=proxy_response.status_code,
            headers=proxy_response_headers,
            background=tasks,
        )

    @override
    async def proxy(*_: Any, **__: Any) -> NoReturn:
        """NotImplemented."""
        raise NotImplementedError()

`proxy(*_, **__)` `async` ¶

NotImplemented.

Source code in src/fastapi_proxy_lib/core/http.py

@override
async def proxy(*_: Any, **__: Any) -> NoReturn:
    """NotImplemented."""
    raise NotImplementedError()

`send_request_to_target(*, request, target_url)` `async` ¶

Change request headers and send request to target url.

The http version of request must be in SUPPORTED_HTTP_VERSIONS.

Parameters:

Name	Type	Description	Default
`request`	`Request`	the original client request.	required
`target_url`	`URL`	target url that request will be sent to.	required

Returns:

Type	Description
`Response`	The response from target url.

Source code in src/fastapi_proxy_lib/core/http.py

@override
async def send_request_to_target(  # pyright: ignore [reportIncompatibleMethodOverride]
    self, *, request: StarletteRequest, target_url: httpx.URL
) -> StarletteResponse:
    """Change request headers and send request to target url.

    - The http version of request must be in [`SUPPORTED_HTTP_VERSIONS`][fastapi_proxy_lib.core.http.SUPPORTED_HTTP_VERSIONS].

    Args:
        request: the original client request.
        target_url: target url that request will be sent to.

    Returns:
        The response from target url.
    """
    client = self.client
    follow_redirects = self.follow_redirects

    check_result = check_http_version(request.scope, SUPPORTED_HTTP_VERSIONS)
    if check_result is not None:
        return check_result

    # 将请求头中的host字段改为目标url的host
    # 同时强制移除"keep-alive"字段和添加"keep-alive"值到"connection"字段中保持连接
    require_close, proxy_header = _change_client_header(
        headers=request.headers, target_url=target_url
    )

    # 有些方法不应该包含主体
    request_content = (
        None if request.method in _NON_REQUEST_BODY_METHODS else request.stream()
    )

    # FIX: https://github.com/WSH032/fastapi-proxy-lib/security/advisories/GHSA-7vwr-g6pm-9hc8
    # time cost: 396 ns ± 3.39 ns
    # 由于这不是原子性的操作，所以不保证一定阻止cookie泄漏
    # 一定能保证修复的方法是通过`_tool.change_necessary_client_header_for_httpx`强制指定优先级最高的cookie头
    client.cookies.clear()

    # NOTE: 不要在这里catch `client.build_request` 和 `client.send` 的异常，因为通常来说
    # - 反向代理的异常需要报 5xx 错误
    # - 而正向代理的异常需要报 4xx 错误
    proxy_request = client.build_request(
        method=request.method,
        url=target_url,
        params=request.query_params,
        headers=proxy_header,
        content=request_content,  # FIXME: 一个已知问题是，流式响应头包含'transfer-encoding': 'chunked'，但有些服务器会400拒绝这个头
        # cookies=request.cookies,  # NOTE: headers中已有的cookie优先级高，所以这里不需要
    )

    # DEBUG: 用于调试的记录
    logging.debug(
        "HTTP: client:%s ; url:%s ; head:%s",
        request.client,
        proxy_request.url,
        proxy_request.headers,
    )

    proxy_response = await client.send(
        proxy_request,
        stream=True,
        follow_redirects=follow_redirects,
    )

    tasks = BackgroundTasks()
    tasks.add_task(proxy_response.aclose)  # 添加后端任务，使其在响应完后关闭

    # 依据先前客户端的请求，决定是否要添加"connection": "close"头到响应头中以关闭连接
    # https://www.uvicorn.org/server-behavior/#http-headers
    # 如果响应头包含"connection": "close"，uvicorn会自动关闭连接
    proxy_response_headers = _change_server_header(
        headers=proxy_response.headers, require_close=require_close
    )
    return StreamingResponse(
        content=proxy_response.aiter_raw(),
        status_code=proxy_response.status_code,
        headers=proxy_response_headers,
        background=tasks,
    )

`ForwardHttpProxy` ¶

Bases: BaseHttpProxy

Forward http proxy.

Attributes:

Name	Type	Description
`client`	`AsyncClient`	The `httpx.AsyncClient` to send http requests.
`follow_redirects`	`bool`	Whether follow redirects of target server.
`proxy_filter`	`ProxyFilterProto`	Callable Filter, decide whether reject the proxy requests.

# Examples¶

from contextlib import asynccontextmanager
from typing import AsyncIterator

from fastapi import FastAPI
from fastapi_proxy_lib.core.http import ForwardHttpProxy
from fastapi_proxy_lib.core.tool import default_proxy_filter
from httpx import AsyncClient
from starlette.requests import Request

proxy = ForwardHttpProxy(AsyncClient(), proxy_filter=default_proxy_filter)

@asynccontextmanager
async def close_proxy_event(_: FastAPI) -> AsyncIterator[None]:
    """Close proxy."""
    yield
    await proxy.aclose()

app = FastAPI(lifespan=close_proxy_event)

@app.get("/{path:path}")
async def _(request: Request, path: str = ""):
    return await proxy.proxy(request=request, path=path)

# Then run shell: `uvicorn <your.py>:app --host http://127.0.0.1:8000 --port 8000`
# visit the app: `http://127.0.0.1:8000/http://www.example.com`
# you will get the response from `http://www.example.com`

Source code in src/fastapi_proxy_lib/core/http.py

class ForwardHttpProxy(BaseHttpProxy):
    '''Forward http proxy.

    Attributes:
        client: The [`httpx.AsyncClient`](https://www.python-httpx.org/api/#asyncclient) to send http requests.
        follow_redirects: Whether follow redirects of target server.
        proxy_filter: Callable Filter, decide whether reject the proxy requests.

    # # Examples

    ```python
    from contextlib import asynccontextmanager
    from typing import AsyncIterator

    from fastapi import FastAPI
    from fastapi_proxy_lib.core.http import ForwardHttpProxy
    from fastapi_proxy_lib.core.tool import default_proxy_filter
    from httpx import AsyncClient
    from starlette.requests import Request

    proxy = ForwardHttpProxy(AsyncClient(), proxy_filter=default_proxy_filter)

    @asynccontextmanager
    async def close_proxy_event(_: FastAPI) -> AsyncIterator[None]:
        """Close proxy."""
        yield
        await proxy.aclose()

    app = FastAPI(lifespan=close_proxy_event)

    @app.get("/{path:path}")
    async def _(request: Request, path: str = ""):
        return await proxy.proxy(request=request, path=path)

    # Then run shell: `uvicorn <your.py>:app --host http://127.0.0.1:8000 --port 8000`
    # visit the app: `http://127.0.0.1:8000/http://www.example.com`
    # you will get the response from `http://www.example.com`
    ```
    '''

    client: httpx.AsyncClient
    follow_redirects: bool
    proxy_filter: ProxyFilterProto

    @override
    def __init__(
        self,
        client: Optional[httpx.AsyncClient] = None,
        *,
        follow_redirects: bool = False,
        proxy_filter: Optional[ProxyFilterProto] = None,
    ) -> None:
        """Forward http proxy.

        Args:
            client: The `httpx.AsyncClient` to send http requests. Defaults to None.<br>
                If None, will create a new `httpx.AsyncClient`,
                else will use the given `httpx.AsyncClient`.
            follow_redirects: Whether follow redirects of target server. Defaults to False.
            proxy_filter: Callable Filter, decide whether reject the proxy requests.
                If None, will use the default filter.
        """
        # TODO: 当前显式发出警告是有意设计，后续会取消警告
        self.proxy_filter = warn_for_none_filter(proxy_filter)
        super().__init__(client, follow_redirects=follow_redirects)

    @override
    async def proxy(  # pyright: ignore [reportIncompatibleMethodOverride]
        self,
        *,
        request: StarletteRequest,
        path: Optional[str] = None,
    ) -> StarletteResponse:
        """Send request to target server.

        Args:
            request: `starlette.requests.Request`
            path: The path params of request, which means the full url of target server.<br>
                If None, will get it from `request.path_params`.<br>
                **Usually, you don't need to pass this argument**.

        Returns:
            The response from target server.
        """
        proxy_filter = self.proxy_filter

        # 只取第一个路径参数
        path_param: str = (
            next(iter(request.path_params.values()), "") if path is None else path
        )
        # 如果没有路径参数，即在正向代理中未指定目标url，则返回400
        if path_param == "":
            error = _BadTargetUrlError("Must provide target url.")
            return return_err_msg_response(
                error, status_code=starlette_status.HTTP_400_BAD_REQUEST
            )

        # 尝试解析路径参数为url
        try:
            # NOTE: 在前向代理中，路径参数即为目标url。
            # TODO: 每次实例化URL都要消耗16.2 µs，考虑是否用lru_cache来优化
            target_url = httpx.URL(path_param)
        except httpx.InvalidURL as e:  # pragma: no cover
            # 这个错误应该是不会被引起的，因为接收到的path_param是经过校验的
            # 但不排除有浏览器不遵守最大url长度限制，发出了超长的url导致InvalidURL错误
            # 所以我们在这里记录这个严重错误，表示要去除 `pragma: no cover`
            return return_err_msg_response(
                e,
                status_code=starlette_status.HTTP_400_BAD_REQUEST,
                logger=logging.critical,
            )

        # 进行请求过滤
        filter_result = proxy_filter(target_url)
        if filter_result is not None:
            return return_err_msg_response(
                _RejectedProxyRequestError(filter_result),
                status_code=starlette_status.HTTP_403_FORBIDDEN,
            )

        try:
            return await self.send_request_to_target(
                request=request, target_url=target_url
            )
        # 需要检查客户端输入的url是否合法，包括缺少scheme; 如果不合符会引发 _400 异常
        except _400_ERROR_NEED_TO_BE_CATCHED_IN_FORWARD_PROXY as e:
            return return_err_msg_response(
                e, status_code=starlette_status.HTTP_400_BAD_REQUEST
            )
        except _500_ERROR_NEED_TO_BE_CATCHED_IN_FORWARD_PROXY as e:
            # 5xx 错误需要记录
            return return_err_msg_response(
                e,
                status_code=starlette_status.HTTP_500_INTERNAL_SERVER_ERROR,
                logger=logging.exception,
                _exc_info=e,
            )

`init(client=None, *, follow_redirects=False, proxy_filter=None)` ¶

Forward http proxy.

Parameters:

Name	Type	Description	Default
`client`	`Optional[AsyncClient]`	The `httpx.AsyncClient` to send http requests. Defaults to None. If None, will create a new `httpx.AsyncClient`, else will use the given `httpx.AsyncClient`.	`None`
`follow_redirects`	`bool`	Whether follow redirects of target server. Defaults to False.	`False`
`proxy_filter`	`Optional[ProxyFilterProto]`	Callable Filter, decide whether reject the proxy requests. If None, will use the default filter.	`None`

Source code in src/fastapi_proxy_lib/core/http.py

@override
def __init__(
    self,
    client: Optional[httpx.AsyncClient] = None,
    *,
    follow_redirects: bool = False,
    proxy_filter: Optional[ProxyFilterProto] = None,
) -> None:
    """Forward http proxy.

    Args:
        client: The `httpx.AsyncClient` to send http requests. Defaults to None.<br>
            If None, will create a new `httpx.AsyncClient`,
            else will use the given `httpx.AsyncClient`.
        follow_redirects: Whether follow redirects of target server. Defaults to False.
        proxy_filter: Callable Filter, decide whether reject the proxy requests.
            If None, will use the default filter.
    """
    # TODO: 当前显式发出警告是有意设计，后续会取消警告
    self.proxy_filter = warn_for_none_filter(proxy_filter)
    super().__init__(client, follow_redirects=follow_redirects)

`proxy(*, request, path=None)` `async` ¶

Send request to target server.

Parameters:

Name	Type	Description	Default
`request`	`Request`	`starlette.requests.Request`	required
`path`	`Optional[str]`	The path params of request, which means the full url of target server. If None, will get it from `request.path_params`. Usually, you don't need to pass this argument.	`None`

Returns:

Type	Description
`Response`	The response from target server.

Source code in src/fastapi_proxy_lib/core/http.py

@override
async def proxy(  # pyright: ignore [reportIncompatibleMethodOverride]
    self,
    *,
    request: StarletteRequest,
    path: Optional[str] = None,
) -> StarletteResponse:
    """Send request to target server.

    Args:
        request: `starlette.requests.Request`
        path: The path params of request, which means the full url of target server.<br>
            If None, will get it from `request.path_params`.<br>
            **Usually, you don't need to pass this argument**.

    Returns:
        The response from target server.
    """
    proxy_filter = self.proxy_filter

    # 只取第一个路径参数
    path_param: str = (
        next(iter(request.path_params.values()), "") if path is None else path
    )
    # 如果没有路径参数，即在正向代理中未指定目标url，则返回400
    if path_param == "":
        error = _BadTargetUrlError("Must provide target url.")
        return return_err_msg_response(
            error, status_code=starlette_status.HTTP_400_BAD_REQUEST
        )

    # 尝试解析路径参数为url
    try:
        # NOTE: 在前向代理中，路径参数即为目标url。
        # TODO: 每次实例化URL都要消耗16.2 µs，考虑是否用lru_cache来优化
        target_url = httpx.URL(path_param)
    except httpx.InvalidURL as e:  # pragma: no cover
        # 这个错误应该是不会被引起的，因为接收到的path_param是经过校验的
        # 但不排除有浏览器不遵守最大url长度限制，发出了超长的url导致InvalidURL错误
        # 所以我们在这里记录这个严重错误，表示要去除 `pragma: no cover`
        return return_err_msg_response(
            e,
            status_code=starlette_status.HTTP_400_BAD_REQUEST,
            logger=logging.critical,
        )

    # 进行请求过滤
    filter_result = proxy_filter(target_url)
    if filter_result is not None:
        return return_err_msg_response(
            _RejectedProxyRequestError(filter_result),
            status_code=starlette_status.HTTP_403_FORBIDDEN,
        )

    try:
        return await self.send_request_to_target(
            request=request, target_url=target_url
        )
    # 需要检查客户端输入的url是否合法，包括缺少scheme; 如果不合符会引发 _400 异常
    except _400_ERROR_NEED_TO_BE_CATCHED_IN_FORWARD_PROXY as e:
        return return_err_msg_response(
            e, status_code=starlette_status.HTTP_400_BAD_REQUEST
        )
    except _500_ERROR_NEED_TO_BE_CATCHED_IN_FORWARD_PROXY as e:
        # 5xx 错误需要记录
        return return_err_msg_response(
            e,
            status_code=starlette_status.HTTP_500_INTERNAL_SERVER_ERROR,
            logger=logging.exception,
            _exc_info=e,
        )

`ReverseHttpProxy` ¶

Bases: BaseHttpProxy

Reverse http proxy.

Attributes:

Name	Type	Description
`client`	`AsyncClient`	The `httpx.AsyncClient` to send http requests.
`follow_redirects`	`bool`	Whether follow redirects of target server.
`base_url`	`URL`	The target server url.

# Examples¶

from contextlib import asynccontextmanager
from typing import AsyncIterator

from fastapi import FastAPI
from fastapi_proxy_lib.core.http import ReverseHttpProxy
from httpx import AsyncClient
from starlette.requests import Request

proxy = ReverseHttpProxy(AsyncClient(), base_url="http://www.example.com/")

@asynccontextmanager
async def close_proxy_event(_: FastAPI) -> AsyncIterator[None]:  # (1)!
    """Close proxy."""
    yield
    await proxy.aclose()

app = FastAPI(lifespan=close_proxy_event)

@app.get("/{path:path}")  # (2)!
async def _(request: Request, path: str = ""):
    return await proxy.proxy(request=request, path=path)  # (3)!

# Then run shell: `uvicorn <your.py>:app --host http://127.0.0.1:8000 --port 8000`
# visit the app: `http://127.0.0.1:8000/`
# you will get the response from `http://www.example.com/`

lifespan please refer to starlette/lifespan
{path:path} is the key.
It allows the app to accept all path parameters.
visit https://www.starlette.io/routing/#path-parameters for more info.
Info
In fact, you only need to pass the request: Request argument.
fastapi_proxy_lib can automatically get the path from request.
Explicitly pointing it out here is just to remind you not to forget to specify {path:path}.

Source code in src/fastapi_proxy_lib/core/http.py

class ReverseHttpProxy(BaseHttpProxy):
    '''Reverse http proxy.

    Attributes:
        client: The [`httpx.AsyncClient`](https://www.python-httpx.org/api/#asyncclient) to send http requests.
        follow_redirects: Whether follow redirects of target server.
        base_url: The target server url.

    # # Examples

    ```python
    from contextlib import asynccontextmanager
    from typing import AsyncIterator

    from fastapi import FastAPI
    from fastapi_proxy_lib.core.http import ReverseHttpProxy
    from httpx import AsyncClient
    from starlette.requests import Request

    proxy = ReverseHttpProxy(AsyncClient(), base_url="http://www.example.com/")

    @asynccontextmanager
    async def close_proxy_event(_: FastAPI) -> AsyncIterator[None]:  # (1)!
        """Close proxy."""
        yield
        await proxy.aclose()

    app = FastAPI(lifespan=close_proxy_event)

    @app.get("/{path:path}")  # (2)!
    async def _(request: Request, path: str = ""):
        return await proxy.proxy(request=request, path=path)  # (3)!

    # Then run shell: `uvicorn <your.py>:app --host http://127.0.0.1:8000 --port 8000`
    # visit the app: `http://127.0.0.1:8000/`
    # you will get the response from `http://www.example.com/`
    ```

    1. lifespan please refer to [starlette/lifespan](https://www.starlette.io/lifespan/)
    2. `{path:path}` is the key.<br>
        It allows the app to accept all path parameters.<br>
        visit <https://www.starlette.io/routing/#path-parameters> for more info.
    3. !!! info
        In fact, you only need to pass the `request: Request` argument.<br>
        `fastapi_proxy_lib` can automatically get the `path` from `request`.<br>
        Explicitly pointing it out here is just to remind you not to forget to specify `{path:path}`.
    '''

    client: httpx.AsyncClient
    follow_redirects: bool
    base_url: httpx.URL

    @override
    def __init__(
        self,
        client: Optional[httpx.AsyncClient] = None,
        *,
        base_url: Union[httpx.URL, str],
        follow_redirects: bool = False,
    ) -> None:
        """Reverse http proxy.

        Note: please make sure `base_url` is available.
            Because when an error occurs,
            we cannot distinguish whether it is a proxy server network error, or it is a error of `base_url`.
            So, we will return 502 status_code whatever the error is.

        Args:
            client: The `httpx.AsyncClient` to send http requests. Defaults to None.<br>
                If None, will create a new `httpx.AsyncClient`,
                else will use the given `httpx.AsyncClient`.
            follow_redirects: Whether follow redirects of target server. Defaults to False.
            base_url: The target proxy server url.
        """
        self.base_url = check_base_url(base_url)
        super().__init__(client, follow_redirects=follow_redirects)

    @override
    async def proxy(  # pyright: ignore [reportIncompatibleMethodOverride]
        self, *, request: StarletteRequest, path: Optional[str] = None
    ) -> StarletteResponse:
        """Send request to target server.

        Args:
            request: `starlette.requests.Request`
            path: The path params of request, which means the path params of base url.<br>
                If None, will get it from `request.path_params`.<br>
                **Usually, you don't need to pass this argument**.

        Returns:
            The response from target server.
        """
        base_url = self.base_url

        # 只取第一个路径参数。注意，我们允许没有路径参数，这代表直接请求
        path_param: str = (
            path if path is not None else next(iter(request.path_params.values()), "")
        )

        # 将路径参数拼接到目标url上
        # e.g: "https://www.example.com/p0/" + "p1"
        # NOTE: 这里的 path_param 是不带查询参数的，且允许以 "/" 开头 (最终为/p0//p1)
        target_url = base_url.copy_with(
            path=(base_url.path + path_param)
        )  # 耗时: 18.4 µs ± 262 ns

        try:
            return await self.send_request_to_target(
                request=request, target_url=target_url
            )
        except _502_ERROR_NEED_TO_BE_CATCHED_IN_REVERSE_PROXY as e:
            # 请注意，反向代理服务器（即此实例）有义务保证:
            # 无论客户端输入的路径参数是什么，代理服务器与上游服务器之间的网络连接始终都应是可用的
            # 因此这里出现任何错误，都认为代理服务器（即此实例）的内部错误，将返回502
            msg = dedent(
                f"""\
                Error in ReverseHttpProxy().proxy():
                url: {target_url}
                request headers: {request.headers}
                """
            )  # 最好不要去查询request.body，因为可能会很大，比如在上传文件的post请求中

            return return_err_msg_response(
                _ReverseProxyServerError(
                    "Oops! Something wrong! Please contact the server maintainer!"
                ),
                status_code=starlette_status.HTTP_502_BAD_GATEWAY,
                logger=logging.exception,
                _msg=msg,
                _exc_info=e,
            )

`init(client=None, *, base_url, follow_redirects=False)` ¶

Reverse http proxy.

please make sure base_url is available.

Because when an error occurs, we cannot distinguish whether it is a proxy server network error, or it is a error of base_url. So, we will return 502 status_code whatever the error is.

Parameters:

Name	Type	Description	Default
`client`	`Optional[AsyncClient]`	The `httpx.AsyncClient` to send http requests. Defaults to None. If None, will create a new `httpx.AsyncClient`, else will use the given `httpx.AsyncClient`.	`None`
`follow_redirects`	`bool`	Whether follow redirects of target server. Defaults to False.	`False`
`base_url`	`Union[URL, str]`	The target proxy server url.	required

Source code in src/fastapi_proxy_lib/core/http.py

@override
def __init__(
    self,
    client: Optional[httpx.AsyncClient] = None,
    *,
    base_url: Union[httpx.URL, str],
    follow_redirects: bool = False,
) -> None:
    """Reverse http proxy.

    Note: please make sure `base_url` is available.
        Because when an error occurs,
        we cannot distinguish whether it is a proxy server network error, or it is a error of `base_url`.
        So, we will return 502 status_code whatever the error is.

    Args:
        client: The `httpx.AsyncClient` to send http requests. Defaults to None.<br>
            If None, will create a new `httpx.AsyncClient`,
            else will use the given `httpx.AsyncClient`.
        follow_redirects: Whether follow redirects of target server. Defaults to False.
        base_url: The target proxy server url.
    """
    self.base_url = check_base_url(base_url)
    super().__init__(client, follow_redirects=follow_redirects)

`proxy(*, request, path=None)` `async` ¶

Send request to target server.

Parameters:

Name	Type	Description	Default
`request`	`Request`	`starlette.requests.Request`	required
`path`	`Optional[str]`	The path params of request, which means the path params of base url. If None, will get it from `request.path_params`. Usually, you don't need to pass this argument.	`None`

Returns:

Type	Description
`Response`	The response from target server.

Source code in src/fastapi_proxy_lib/core/http.py

@override
async def proxy(  # pyright: ignore [reportIncompatibleMethodOverride]
    self, *, request: StarletteRequest, path: Optional[str] = None
) -> StarletteResponse:
    """Send request to target server.

    Args:
        request: `starlette.requests.Request`
        path: The path params of request, which means the path params of base url.<br>
            If None, will get it from `request.path_params`.<br>
            **Usually, you don't need to pass this argument**.

    Returns:
        The response from target server.
    """
    base_url = self.base_url

    # 只取第一个路径参数。注意，我们允许没有路径参数，这代表直接请求
    path_param: str = (
        path if path is not None else next(iter(request.path_params.values()), "")
    )

    # 将路径参数拼接到目标url上
    # e.g: "https://www.example.com/p0/" + "p1"
    # NOTE: 这里的 path_param 是不带查询参数的，且允许以 "/" 开头 (最终为/p0//p1)
    target_url = base_url.copy_with(
        path=(base_url.path + path_param)
    )  # 耗时: 18.4 µs ± 262 ns

    try:
        return await self.send_request_to_target(
            request=request, target_url=target_url
        )
    except _502_ERROR_NEED_TO_BE_CATCHED_IN_REVERSE_PROXY as e:
        # 请注意，反向代理服务器（即此实例）有义务保证:
        # 无论客户端输入的路径参数是什么，代理服务器与上游服务器之间的网络连接始终都应是可用的
        # 因此这里出现任何错误，都认为代理服务器（即此实例）的内部错误，将返回502
        msg = dedent(
            f"""\
            Error in ReverseHttpProxy().proxy():
            url: {target_url}
            request headers: {request.headers}
            """
        )  # 最好不要去查询request.body，因为可能会很大，比如在上传文件的post请求中

        return return_err_msg_response(
            _ReverseProxyServerError(
                "Oops! Something wrong! Please contact the server maintainer!"
            ),
            status_code=starlette_status.HTTP_502_BAD_GATEWAY,
            logger=logging.exception,
            _msg=msg,
            _exc_info=e,
        )

http

SUPPORTED_HTTP_VERSIONS = ('1.0', '1.1') module-attribute ¶

BaseHttpProxy ¶

proxy(*_, **__) async ¶

send_request_to_target(*, request, target_url) async ¶

ForwardHttpProxy ¶

# Examples¶

__init__(client=None, *, follow_redirects=False, proxy_filter=None) ¶

proxy(*, request, path=None) async ¶

ReverseHttpProxy ¶

# Examples¶

__init__(client=None, *, base_url, follow_redirects=False) ¶

proxy(*, request, path=None) async ¶

`SUPPORTED_HTTP_VERSIONS = ('1.0', '1.1')` `module-attribute` ¶

`BaseHttpProxy` ¶

`proxy(*_, **__)` `async` ¶

`send_request_to_target(*, request, target_url)` `async` ¶

`ForwardHttpProxy` ¶

`init(client=None, *, follow_redirects=False, proxy_filter=None)` ¶

`proxy(*, request, path=None)` `async` ¶

`ReverseHttpProxy` ¶

`init(client=None, *, base_url, follow_redirects=False)` ¶

`proxy(*, request, path=None)` `async` ¶