Refactor basic auth header building

httpx/middleware/basic_auth.py

@@ -2,26 +2,26 @@
 from base64 import b64encode
 
 from ..models import AsyncRequest, AsyncResponse
+from ..utils import to_bytes
 from .base import BaseMiddleware
 
 
 class BasicAuthMiddleware(BaseMiddleware):
     def __init__(
         self, username: typing.Union[str, bytes], password: typing.Union[str, bytes]
     ):
-        if isinstance(username, str):
-            username = username.encode("latin1")
-
-        if isinstance(password, str):
-            password = password.encode("latin1")
-
-        userpass = b":".join((username, password))
-        token = b64encode(userpass).decode().strip()
-
-        self.authorization_header = f"Basic {token}"
+        self.authorization_header = build_basic_auth_header(username, password)
 
     async def __call__(
         self, request: AsyncRequest, get_response: typing.Callable
     ) -> AsyncResponse:
         request.headers["Authorization"] = self.authorization_header
         return await get_response(request)
+
+
+def build_basic_auth_header(
+    username: typing.Union[str, bytes], password: typing.Union[str, bytes]
+) -> str:
+    userpass = b":".join((to_bytes(username), to_bytes(password)))
+    token = b64encode(userpass).decode().strip()
+    return f"Basic {token}"

httpx/utils.py

@@ -169,3 +169,7 @@ def get_logger(name: str) -> logging.Logger:
             logger.addHandler(handler)
 
     return logging.getLogger(name)
+
+
+def to_bytes(value: typing.Union[str, bytes]) -> bytes:
+    return value.encode() if isinstance(value, str) else value