Django Channels Postgres InterfaceError:连接已关闭

2024-05-12

我似乎无法理解这里的问题。我正在为我的渠道消费者编写测试文档中的描述 https://channels.readthedocs.io/en/latest/topics/testing.html。我通常会使用 Django 默认的单元测试,但由于 Channels 需要使用 pytest,我正在使用它,但仍然希望维护编写测试的类结构。

我在编写类似于 unittest 的设置方法时遇到问题,我将使用它来初始化测试属性。我试过方法夹具 https://docs.pytest.org/en/latest/xunit_setup.html?highlight=setup_class#method-and-function-level-setup-teardown,类灯具使用自动使用=真 https://docs.pytest.org/en/latest/fixture.html#autouse-fixtures-xunit-setup-on-steroids,但它们似乎都不起作用,因为我无法访问测试方法中的实例属性。最后我决定写set_up实例方法并为每个类手动调用它。这是我到目前为止所拥有的:

@pytest.mark.django_db
@pytest.mark.asyncio
class TestChatConsumer(object):

    @database_sync_to_async
    def set_up(self):

        self.user = UserFactory()
        self.api_client = APIClientFactory()
        self.token = TokenFactory(user=self.user)

        self.credentials = {
            'AUTHORIZATION': self.token.key,
            ...
        }

        self.credentials_params = '&'.join(['{}={}'.format(k, v) for k, v in self.credentials.items()])
        self.authless_endpoint = '/api/v1/ws/'
        self.endpoint = self.authless_endpoint + '?' + self.credentials_params

    async def test_connect_without_credentials(self):
        await self.set_up()
        communicator = WebsocketCommunicator(application, self.authless_endpoint)
        connected, subprotocol = await communicator.connect()
        assert not connected

    async def test_connect_with_credentials(self):
        await self.set_up()
        communicator = WebsocketCommunicator(application, self.endpoint)
        connected, subprotocol = await communicator.connect()
        assert connected

问题是我不断得到psycopg2.InterfaceError: connection already closed当我的消费者尝试访问数据库中的代码时connect方法。它用database_sync_to_async当我手动测试它时效果很好。

具体来说,该行中connect引发错误的方法如下所示:

await database_sync_to_async(user.inbox.increment_connections)().

不确定问题到底是什么database_sync_to_async应正确清理旧连接,以便可以正确创建新连接。

任何帮助将非常感激。


显然我看错地方了。问题就在这条线上:

await database_sync_to_async(user.inbox.increment_connections)().

inbox是一个相关的管理器,因此它会尝试在实际执行之前获取它database_sync_to_async它失败了,因为它需要数据库调用。

我试过这个await database_sync_to_async(user).inbox.increment_connections()它不起作用,因为它包装了一个方法,而不是一个属性,所以我最终做的是使用prefetch_related得到inbox认证过程中。以这种方式user.inbox不再需要数据库调用并且工作正常

本文内容由网友自发贡献,版权归原作者所有,本站不承担相应法律责任。如您发现有涉嫌抄袭侵权的内容,请联系:hwhale#tublm.com(使用前将#替换为@)

Django Channels Postgres InterfaceError:连接已关闭 的相关文章

  • 在 Django 中保存模型会出现“警告:字段‘id’没有默认值”

    我在 Django 中有一个非常基本的模型 class Case models Model name models CharField max length 255 created at models DateTimeField defau
  • 限制 Django 中的暴力登录攻击 [关闭]

    Closed 这个问题需要多问focused help closed questions 目前不接受答案 是否有普遍接受的策略来保护 Django 应用程序免受此类攻击 有很多可用的库 例如Django 斧头 Django 防御者 Djan
  • Django 中基于类的视图分页

    我正在尝试对基于类的视图进行分页 这是我的观点 class IssuesByTitleView ListView context object name issue list def issues request issue list Is
  • 从 asyncio 子进程获取实时输出

    我正在尝试使用 Python asyncio 子进程来启动交互式 SSH 会话并自动输入密码 实际用例并不重要 但它有助于说明我的问题 这是我的代码 proc await asyncio create subprocess exec ssh
  • 如何去除html源代码中的空格

    我正在使用 django 和 python 在模板文件中 我有一个下拉列表 如下所示 有用 唯一的问题是源 html 代码之间有很多空白 有什么办法可以去除空白吗 谢谢 for lang ele in video languages all
  • python:日志记录:我们可以向记录器添加多个过滤器吗?考虑哪一个

    我试图了解 Python 日志记录中的多个过滤器 一个在配置中定义 另一个在代码中定义 如何工作 我正在开发一个 Django 项目 下面是我在 settings py 中的记录器配置 我的目标是switch on and switch o
  • 如何在 Django 模板中显示站点的用户友好名称?

    我正在寻找一种在模板中显示我的 Django CMS 站点名称的方法 我创建了一个基本的 Django CMS 网站如下本指南 http docs django cms org en release 3 4 x introduction i
  • 基于列顺序的查询速度

    数据库中列类型的顺序对查询时间有影响吗 例如 具有混合排序 INT TEXT VARCHAR INT TEXT 的表的查询速度是否会比具有连续类型 INT INT VARCHAR TEXT TEXT 的表慢 答案是肯定的 这确实很重要 而且
  • 返回行位置 - Postgres

    我返回一个带有位置的表 select from select row number over as position from organization result where data1 Hello 返回这个 这是正确的 data1 H
  • django 中计数器的原子增量

    我正在尝试自动增加 Django 中的一个简单计数器 我的代码如下所示 from models import Counter from django db import transaction transaction commit on s
  • django 根据 UserProfile.language() 字段数据发送本地化电子邮件

    在我的优惠网站中 如果优惠满足某些要求 模型中指定的过滤器 用户可以设置电子邮件警报 因此 当用户 A 添加报价时 post save 信号将发送到 celery 并检查是否应用了用户警报过滤器 如果有 则发送电子邮件 问题是我不知道如何安
  • Postgresql 中的 id 列位置重要吗?

    我正在测试删除主键列 id 的迁移 我想使用外键作为主键 当我运行并恢复迁移时 我看到表的状态是相同的 只是 id 列现在是最后一个 它会以任何方式改变我的数据库的行为吗 我是否应该费心去恢复迁移恢复代码中的列顺序 理论上一切都应该没问题
  • 从备份恢复 PostgreSQL 数据库,没有外键约束问题

    我有一个包含大约 85 个以上表的 postgresql 数据库 我定期使用pg dump 通过 php pgadmin 在复制模式下 备份文件的大小几乎为 10 12 MB 现在我面临的问题是 每当我尝试恢复数据库时 都会出现外键约束问题
  • 为什么最简单的 requests_mock 示例在 pytest 中失败?

    我有一个特殊的问题requests mock 我想用它pytest测试我的 API 包装器库 我尝试过使用requests mock 文档中的第一个示例 http requests mock readthedocs io en latest
  • 如何获取Django当前的语言?

    如何获取 Django 中的当前语言 特别感兴趣的功能是django utils translation get language 它返回当前线程中使用的语言 看文档 https docs djangoproject com en dev
  • 在 Django Rest Framework 中访问视图集和序列化器中的请求对象?

    我想访问 DRF 中 Views py 和 Serializers py 中的请求对象 我的观点 py class ProductViewSet viewsets ReadOnlyModelViewSet This viewset auto
  • Rails csv 格式的原始查询,通过控制器返回

    我使用 Active Record 来获取我的故事 然后生成 CSV 这是在 Rails Cast 中完成的标准方法 但我有很多行 需要几分钟 我想如果我能让 posgresql 来做 csv 渲染 那么我可以节省一些时间 这是我现在所拥有
  • 在 Postgres 中的数组字段上应用聚合函数?

    是否可以对整数 字段 或其他数字数组 中的所有值应用聚合 如 avg stddev CREATE TABLE widget measurement integer insert into widget measurement values
  • Postgresql插入触发器设置值

    假设在 Postgresql 中 我有一个表T它的一栏是C1 我想在新记录添加到表中时触发一个函数T 该函数应该检查列的值C1在新记录中 如果它为 null 空 则将其值设置为 X 这可能吗 您需要触发器是正确的 因为为列设置默认值对您不起
  • 如何避免应用程序停止后 docker 容器停止

    有一个带有 Postgres 服务器的 docker 容器 postgres 停止或崩溃 无关紧要 我需要检查一些环境变量和一些文件的状态 默认情况下 容器在应用程序完成后停止 我知道有一个选项可以更改 dockerfile 中的默认行为

随机推荐