Pysolr传递JWT Token作为认证参数时报TypeError错误求助

开发者阿泽

2026-4-15

大家好，我碰到了一个用pysolr结合JWT认证访问SolrCloud的问题，折腾了好久没解决，想请大家帮忙看看。

环境信息

SOLR v. 9.8.0，3节点集群
Zookeeper v. 3.9.3，3实例
Pysolr最新版本
SOLR和Zookeeper运行在Docker环境中
SOLR认证使用MultiAuthPlugin：BasicAuthPlugin用于Admin UI登录，JWTAuthPlugin用于API请求认证

需求

我需要从FastAPI的RestAPI发送请求到SOLR执行搜索，使用pysolr库实现。

问题

当我把生成的JWT token直接作为auth参数传给pysolr.SolrCloud时，总是抛出TypeError: 'str' object is not callable错误。

已尝试的操作

不使用认证时，搜索功能正常，能返回预期结果；
使用静态token和动态生成的token都会触发相同错误；
把生成的token复制到Postman中发送请求，能正常访问SOLR并返回结果；
使用Basic认证（用户名+密码）时，pysolr能正常工作。

预期结果

请求能正常发送到SOLR，SOLR通过security.json中的配置验证JWT token有效性，最终返回搜索结果。

代码片段

from typing import Any

import jwt
import pysolr
from fastapi import APIRouter, Depends, HTTPException, status
from fastapi_cache.decorator import cache

from app.caching import TTL
from app.core.config import settings
from app.models import SearchResults
from app.queryparams import SearchQueryParams
from app.utils import check_search_input

router = APIRouter()

@router.get("/", response_model=SearchResults)
# @cache(expire=TTL.WEEK, namespace="public:search")
async def search(
    skip: int = 0,
    limit: int = 100,
    params: SearchQueryParams = Depends(),
) -> Any:
    """
    Search query.
    """
    if not check_search_input(params=params):
        raise HTTPException(
            status_code=status.HTTP_422_UNPROCESSABLE_ENTITY,
            detail="Mindestens ein Feld muss ausgefüllt werden",
        )

    zookeeper = pysolr.ZooKeeper("search-zoo1,search-zoo2,search-zoo3")

    jwt_token = jwt.encode({
        "sub": "test-backend",
        "name": "test-backend",
        "iat": 1516239022,
        "admin": False,
        "iss": "test-solr-auth",
        "solr": "test-backend",
        "aud": "test-solr-api",
        "exp": 33292598400,
        "scope": "test-backend",
        "realm": "test-solr-jwt"
    }, settings.KEY, algorithm="RS256")

    solr = pysolr.SolrCloud(zookeeper, "tag", auth=jwt_token)

    res = solr.search(q=params.query, start=skip, rows=limit)

    return SearchResults(data=res, count=res.hits)

错误栈

INFO: "GET /v1/search/?skip=0&limit=100&query=title%3A%22Yellow%22&descending=false HTTP/1.1" 500 Internal Server Error
ERROR:    Exception in ASGI application
Traceback (most recent call last):
   File "/app/.venv/lib/python3.13/site-packages/uvicorn/protocols/http/httptools_impl.py", line 401, in run_asgi
     result = await app(  # type: ignore[func-returns-value]
              ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
         self.scope, self.receive, self.send
         ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
     )
     ^
   File "/app/.venv/lib/python3.13/site-packages/uvicorn/middleware/proxy_headers.py", line 60, in __call__
     return await self.app(scope, receive, send)
            ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
   File "/app/.venv/lib/python3.13/site-packages/fastapi/applications.py", line 1054, in __call__
     await super().__call__(scope, receive, send)
   File "/app/.venv/lib/python3.13/site-packages/starlette/applications.py", line 113, in __call__
     await self.middleware_stack(scope, receive, send)
   File "/app/.venv/lib/python3.13/site-packages/starlette/middleware/errors.py", line 187, in __call__
     raise exc
   File "/app/.venv/lib/python3.13/site-packages/starlette/middleware/errors.py", line 165, in __call__
     await self.app(scope, receive, _send)
   File "/app/.venv/lib/python3.13/site-packages/starlette/middleware/cors.py", line 93, in __call__
     await self.simple_response(scope, receive, send, request_headers=headers)
   File "/app/.venv/lib/python3.13/site-packages/starlette/middleware/cors.py", line 144, in simple_response
     await self.app(scope, receive, send)
   File "/app/.venv/lib/python3.13/site-packages/starlette/middleware/exceptions.py", line 62, in __call__
     await wrap_app_handling_exceptions(self.app, conn)(scope, receive, send)
   File "/app/.venv/lib/python3.13/site-packages/starlette/_exception_handler.py", line 53, in wrapped_app
     raise exc
   File "/app/.venv/lib/python3.13/site-packages/starlette/_exception_handler.py", line 42, in wrapped_app
     await app(scope, receive, sender)
   File "/app/.venv/lib/python3.13/site-packages/starlette/routing.py", line 715, in __call__
     await self.middleware_stack(scope, receive, send)
   File "/app/.venv/lib/python3.13/site-packages/starlette/routing.py", line 735, in app
     await route.handle(scope, receive, send)
   File "/app/.venv/lib/python3.13/site-packages/starlette/routing.py", line 288, in handle
     await self.app(scope, receive, send)
   File "/app/.venv/lib/python3.13/site-packages/starlette/routing.py", line 76, in app
     await wrap_app_handling_exceptions(app, request)(scope, receive, send)
   File "/app/.venv/lib/python3.13/site-packages/starlette/_exception_handler.py", line 53, in wrapped_app
     raise exc
   File "/app/.venv/lib/python3.13/site-packages/starlette/_exception_handler.py", line 42, in wrapped_app
     await app(scope, receive, sender)
   File "/app/.venv/lib/python3.13/site-packages/starlette/routing.py", line 73, in app
     response = await f(request)
                ^^^^^^^^^^^^^^^^
   File "/app/.venv/lib/python3.13/site-packages/fastapi/routing.py", line 301, in app
     raw_response = await run_endpoint_function(
                    ^^^^^^^^^^^^^^^^^^^^^^^^^^^^
     ...<3 lines>...
     )
     ^
   File "/app/.venv/lib/python3.13/site-packages/fastapi/routing.py", line 212, in run_endpoint_function
     return await dependant.call(**values)
            ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
   File "/app/app/api/routes/public/search.py", line 56, in search
     res = solr.search(q=params.query, start=skip, rows=limit)
   File "/app/.venv/lib/python3.13/site-packages/pysolr.py", line 834, in search
     response = self._select(params, handler=search_handler)
   File "/app/.venv/lib/python3.13/site-packages/pysolr.py", line 486, in _select
     return self._send_request("get", path)
            ~~~~~~~~~~~~~~~~~~^^^^^^^^^^^^^
   File "/app/.venv/lib/python3.13/site-packages/pysolr.py", line 1512, in _send_request
     return Solr._send_request(self, method, path, body, headers, files)
            ~~~~~~~~~~~~~~~~~~^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
   File "/app/.venv/lib/python3.13/site-packages/pysolr.py", line 414, in _send_request
     resp = requests_method(
         url,
     ...<4 lines>...
         auth=self.auth,
     )
   File "/app/.venv/lib/python3.13/site-packages/requests/sessions.py", line 602, in get
     return self.request("GET", url, **kwargs)
            ~~~~~~~~~~~~^^^^^^^^^^^^^^^^^^^^^^
   File "/app/.venv/lib/python3.13/site-packages/requests/sessions.py", line 575, in request
     prep = self.prepare_request(req)
   File "/app/.venv/lib/python3.13/site-packages/requests/sessions.py", line 484, in prepare_request
     p.prepare(
     ~~~~~~~~~^
         method=request.method.upper(),
         ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
     ...<10 lines>...
         hooks=merge_hooks(request.hooks, self.hooks),
         ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
     )
     ^
   File "/app/.venv/lib/python3.13/site-packages/requests/models.py", line 371, in prepare
     self.prepare_auth(auth, url)
     ~~~~~~~~~~~~~~~~~^^^^^^^^^^^
   File "/app/.venv/lib/python3.13/site-packages/requests/models.py", line 602, in prepare_auth
     r = auth(self)
 TypeError: 'str' object is not callable

问题原因分析

从错误栈能看到，问题出在requests库的prepare_auth方法里——它尝试把auth参数当作可调用对象来执行（r = auth(self)），但我们直接传了字符串类型的JWT token，自然会报“str不可调用”的错误。

pysolr的auth参数是直接传递给requests库的，而requests的auth参数只接受两种类型：

一个包含(username, password)的元组（用于Basic认证，这也是为什么用用户名密码能正常工作）；
实现了__call__方法的认证类实例（比如requests.auth.HTTPBasicAuth）。

JWT认证不需要这种“可调用的auth对象”，而是要把token放在请求头的Authorization字段里，格式是Bearer <token>。

解决方案

我们不需要给pysolr传auth参数，而是直接自定义请求头，把JWT token加进去。pysolr的SolrCloud和Solr类支持通过headers参数设置全局请求头，或者在每次请求时单独指定。

方法一：设置全局请求头（所有请求都带这个token）

# 生成token后，构造headers
headers = {"Authorization": f"Bearer {jwt_token}"}

# 初始化SolrCloud时传入headers参数，而不是auth参数
solr = pysolr.SolrCloud(zookeeper, "tag", headers=headers)

方法二：在单个请求中指定headers（适合不同请求用不同token的场景）

如果接口需要根据不同请求生成不同token，可以在调用search方法时传入headers：

# 初始化SolrCloud时不指定auth或headers
solr = pysolr.SolrCloud(zookeeper, "tag")

# 生成token后构造headers
headers = {"Authorization": f"Bearer {jwt_token}"}

# 调用search时传入headers参数
res = solr.search(q=params.query, start=skip, rows=limit, headers=headers)

这样修改后，请求就会带上正确的Authorization头，Solr的JWTAuthPlugin就能正常验证token了。

验证建议

可以先打印一下生成的headers内容，确认格式是Bearer <你的token>，然后再发送请求，应该就能正常返回搜索结果了。

备注：内容来源于stack exchange，提问作者anvbis

Pysolr传递JWT Token作为认证参数时报TypeError错误求助

环境信息 #

需求 #

问题 #

已尝试的操作 #

预期结果 #

代码片段 #

错误栈 #

问题原因分析 #

解决方案 #

方法一：设置全局请求头（所有请求都带这个token）

方法二：在单个请求中指定headers（适合不同请求用不同token的场景）

验证建议 #

环境信息

需求

问题

已尝试的操作

预期结果

代码片段

错误栈

问题原因分析

解决方案

验证建议