如何在python中使用asyncio和postgres进行事务处理？
在Python中使用asyncio和PostgreSQL进行事务处理的步骤如下：
- 导入必要的模块和库：import asyncio import asyncpg
- 创建一个异步函数来处理事务：async def execute_transaction(): conn = await asyncpg.connect(user='your_username', password='your_password', database='your_database', host='your_host') try: async with conn.transaction(): # 在这里执行事务操作 await conn.execute('SQL语句') await conn.execute('SQL语句') # ... finally: await conn.close()
- 在主程序中运行异步函数：loop = asyncio.get_event_loop() loop.run_until_complete(execute_transaction())
在上述代码中，你需要替换以下参数：
your_username：你的PostgreSQL用户名your_password：你的PostgreSQL密码your_database：你要连接的数据库名称your_host：你的PostgreSQL主机地址
这段代码使用了asyncpg库来连接和操作PostgreSQL数据库。在事务处理中，你可以使用conn.execute()方法来执行SQL语句。
关于asyncpg库的更多信息和使用方法，你可以参考腾讯云的PostgreSQL产品文档：asyncpg库使用指南。
需要注意的是，以上代码只是一个简单的示例，实际使用时你可能需要根据具体的业务需求进行适当的修改和扩展。
相关·内容
我有一个python应用程序，它与postgresql数据库交互，我需要在一个docker容器中运行它。在运行容器时，我会得到一个连接错误：
...
File "/usr/lib/python3.6/asyncio/tasks.py", line 358, in wait_for
return fut.result()
File "/usr/lib/python3.6/asyncio/base_events.py", line 787, in create_connection
', '.join(str(exc)
浏览 0提问于2019-01-16得票数 1
1回答
我正在尝试将asyncpg与discord.py集成，但是我遇到了一个非常棘手的问题。
每当我尝试使用^C停止机器人时，我都会收到大量错误的垃圾邮件。这很麻烦，因为当我尝试调试某些东西时，我经常会丢失原始错误。
下面是我的代码：
loop = asyncio.get_event_loop()
async def connection_init(conn):
await conn.execute("SET CLIENT_ENCODING to 'utf-8';")
conn.client = client
try:
client.poo
浏览 15提问于2021-01-22得票数 0
在Postgres后端使用Pytest 6.2.5和SQLAlchemy 1.4异步引擎。
我有两个测试，测试简单的更新和读取函数。它们是SQLAlchemy asyncio方法。update测试更新数据库中的值，read函数断言存在新写入的值。目前，这是两个独立的测试，都通过了。我想把它们变成一个单一的测试，因为它们是相互依存的。
将这两个函数放在一个测试中会导致测试失败。我可以看到它与异步功能相关，但我认为@pytest.mark.asyncio装饰器和asyncio.run()异步特定的功能可以解释这一点。
这两项测试通过：
@pytest.mark.asyncio
def test_u
浏览 6提问于2021-12-04得票数 1
回答已采纳
我正在尝试使用一个单独的进程来通过并发期货来流数据。然而，在另一边，有时对方会停止数据处理。但是，只要我重新启动这个threadable，它就会再次工作。所以我设计了这样的东西，可以在不干预的情况下保持流数据。
executor = concurrent.futures.ProcessPoolExecutor()
job2 = executor.submit(threadable,list_tmp_replace)
time.sleep(3600)
executor_tmp = executor
executor = concurrent.futures.ProcessPoolExecutor
浏览 12提问于2022-03-22得票数 3
回答已采纳
我有一个API，这需要很长时间。因此，我想把它分成许多较小的作业，然后并行地运行它们，然后等待结果，然后发送响应。
我的狙击代码：
@app.post("/data")
async def get_tsp_events:
query = [foo, bar, foo, bar]
tasks = [asyncio.create_task(do_work(param1, param2)) for query in queries]
events = await asyncio.gather(*tasks)
return events
async d
浏览 5提问于2021-07-13得票数 0
在FastAPI上运行测试时，我试图模仿Django的行为:我希望在每个测试开始时创建一个测试数据库，并在最后销毁它。问题是FastAPI的异步性破坏了一切。当我做了一次理智的检查，并把所有的同步，一切都运行得很好。不过，当我尝试运行things异步时，一切都会中断。我现在拥有的是：
固定装置：
@pytest.fixture(scope="session")
def event_loop():
return asyncio.get_event_loop()
@pytest.fixture(scope="session")
async def ses
浏览 18提问于2022-01-21得票数 1
1回答
从SQLAlchemy版本1.4.0b2升级到1.4.0b3时，试图使用异步扩展连接到SQLite引擎时会出现以下错误。
>>> from sqlalchemy.ext.asyncio import create_async_engine
>>> engine = create_async_engine('sqlite:///database.db', echo=True, future=True)
Traceback (most recent call last):
File "<stdin>", line
浏览 1提问于2021-03-01得票数 4
这对于我如何在python3中使用异步非常清楚。
import asyncio
import time
async def do_something(number):
print(f"do something no.{number} at timestamp : {time.perf_counter():6.2f}")
await asyncio.sleep(1)
print(f"do something no.{number} ended at timestamp: {time.perf_counter():6.2f}")
浏览 3提问于2021-05-17得票数 2
回答已采纳
1回答
无法连接异步的Psycopg3
、、、、
无法使用异步psycopg3连接postgres数据库：
import asyncio
import psycopg
async def main():
async with await psycopg.AsyncConnection.connect('postgresql://xxxxxxxxxxx') as con:
async with con.cursor() as cur:
print(await cur.execute('select 1 a').fetchall())
if __name__ ==
浏览 6提问于2022-02-22得票数 0
我练习，这是django测试部分。我的环境是：
Django 4.0
Python 3.9
数据库PostgreSQL 14.2
码头组成
此外，到PostgreSQL的连接通过.pg_service.conf和.pgpass配置。
docker-compose.yml
version: "3.9"
services:
web:
build:
context: .
container_name: web
restart: unless-stopped
tty: true
working_dir: /
浏览 3提问于2022-05-05得票数 2
1回答
我有三个码头集装箱在同一个网络中运行。我用船坞构图把集装箱拿来。码头撰写脚本是：
version: '3.5'
services:
### Jesse's Workspace ################################################
jesse:
image: salehmir/jesse:latest
depends_on:
- postgres
- redis
tty: true
env_file:
- ../.env
ports:
浏览 1提问于2022-08-21得票数 0
我有一个大型项目，它依赖于一些第三方库，有时它的执行会被CancelledError中断。
为了演示这个问题，让我们看一个小例子：
import asyncio
async def main():
task = asyncio.create_task(foo())
# Cancel the task in 1 second.
loop = asyncio.get_event_loop()
loop.call_later(1.0, lambda: task.cancel())
await task
async def foo():
aw
浏览 3提问于2022-03-02得票数 0
回答已采纳
我的docker-compose文件如下 version: "3"
volumes:
postgres:
driver: local
services:
postgres:
image: postgres:11.1
environment:
- POSTGRES_NAME=postgres
- POSTGRES_USER=postgres
- POSTGRES_PASSWORD=mysecretpassword
- PGDATA=pgdata
volumes:
- post
浏览 25提问于2020-07-17得票数 0
这段代码(在线程中运行的最小服务器，取自的代码)在Python3.8.3中工作得很好，但在Python3.9.0中引发了一条错误消息：
import asyncio
import threading
from aiohttp import web
def aiohttp_server():
def say_hello(request):
return web.Response(text='Hello, world')
app = web.Application()
app.add_routes([web.get('/'
浏览 1提问于2020-12-27得票数 1
我在网上搜索大量的地址，我想在我的任务中同时使用异步和ProcessPoolExecutor来快速搜索地址。
async def main():
n_jobs = 3
addresses = [list of addresses]
_addresses = list_splitter(data=addresses, n=n_jobs)
with ProcessPoolExecutor(max_workers=n_jobs) as executor:
futures_list = []
浏览 0提问于2019-05-23得票数 4
回答已采纳
我想在码头容器中设置一个带有PostgresDb的Django应用程序，所以我想要设置docker-compose，但是当我执行代码对接时，django和db都很好，我还开发了一些API，它们也正常工作，但不幸的是，突然之间，这些错误阻止了我：
pgdb_1 |
pgdb_1 | 2021-12-11 15:05:38.674 UTC [1] LOG: starting PostgreSQL 14.1 (Debian 14.1-1.pgdg110+1) on x86_64-pc-linux-gnu, compiled by gcc (Debian 10.2.1-6
浏览 15提问于2021-12-11得票数 0
我目前正在Python3.7中使用asyncio，并使用asyncio.start_server()函数编写了一个TCP服务器。请参阅此示例：
也可以尝试使用“I/O完成端口”(IOCP)的asyncio.ProactorEventLoop
根据微软官方文档，它提到使用I/O完成端口和预先分配的线程池，但我找不到哪里可以分配线程的数量
我可以在哪里分配线程池中的线程数量？有谁能帮帮我吗？非常感谢!
浏览 2提问于2020-06-14得票数 0
我试图了解postgres如何处理事务的细节；我注意到，在psql上执行以下命令时，txid为518526，而不是像我预期的那样低的0或1。
postgres=# CREATE DATABASE test;
CREATE DATABASE
postgres=# \connect test;
You are now connected to database "test" as user "postgres".
test=# begin;
BEGIN
test=# select txid_current();
txid_current
------------
浏览 0提问于2013-12-15得票数 1
回答已采纳
编辑了我的代码--现在它起作用了，我正试图通过异步NOW连接池从Postgres db获取一些数据。基本上，我的db包含大约100个不同的表(每个城市)，并且我试图尽可能快地在一个帧中收集所有数据。
import pandas as pd
import asyncpg
import asyncio
from time import time
def make_t():
lst = []
# iterator for sql tuple
for i in ['a',
浏览 0提问于2019-02-08得票数 1
1回答
Python并发- CAN总线
、、、、
我正处于开发一个可扩展的基于CAN的硬件测试系统的初期阶段。测试系统的两个主要功能是记录总线上的所有流量并执行测试用例。测试用例主要由接收和发送CAN消息组成，同时评估消息数据有效载荷的内容。还将有一系列测试用例，这些测试用例必须与串行控制设备集成。
由于日志记录的重要性，我一直在考虑一个基于并发的程序。但我不确定我应该实现哪种形式的并发。我读过关于asyncIO、线程和多处理的文章。所以我对他们每个人都有一个高层次的看法。
我还在使用python库，这个库似乎内置了asyncIO，因此我相信asyncIO应该很好。我担心，如果测试用例协同处理不够快，我可能不得不使用asyncIO建立某种缓冲
浏览 2提问于2020-06-17得票数 1
回答已采纳
我在websockets中使用异步，当套接字关闭或超时时，我需要做一些事情：
try:
async for msg in ws:
if msg.type == aiohttp.WSMsgType.TEXT:
print(msg)
except concurrent.futures._base.CancelledError:
print("Cancelled")
pass
except (concurrent.futures._base.TimeoutError, asyncio.exceptions.Timeout
浏览 14提问于2020-07-14得票数 0
回答已采纳
编辑：在遵循xcodz dot的建议后，虽然很有帮助，而且我认为它解决了一个问题，但我在client.py中收到了一个例外。回溯：
Traceback (most recent call last):
File "/workspaces/[repo_name]/resource.py", line 5, in <module>
from discord_components import DiscordComponents, ButtonStyle, Button
File "/opt/python/3.10.4/lib/python3.10
浏览 10提问于2022-06-11得票数 0
回答已采纳
7回答
如何在异步中使用请求？
、、、
我想在asyncio中并行执行http请求任务，但是我发现python-requests会阻塞asyncio的事件循环。我已经找到了，但它无法使用http代理提供http请求服务。
因此，我想知道是否有一种方法可以在asyncio的帮助下执行异步http请求。
浏览 3提问于2014-03-05得票数 191
回答已采纳
我有两个数据库cvtl和cvtl_db，我需要编写一个查询来从cvtl中的表A和cvtl_db中的表B中检索数据。
Postgres抛出错误:未实现跨数据库引用
浏览 0提问于2016-03-30得票数 0
我开始学习Django，我在使用Postgres数据库时遇到了问题，服务器在sqlite3上工作得很好，但是当我在项目中声明Postgres时，服务器停止了工作。
DATABASES = {
'default': {
'ENGINE': 'django.db.backends.postgresql', # on utilise l'adaptateur postgresql
'NAME': 'sarah disquaires', # le nom de notre base de donne
浏览 0提问于2021-09-12得票数 0
1回答
我正在开发一个控制器应用程序，它监视和控制独立的python可执行的子进程。基本上，我想要的是在controller.py中运行一个asyncio.star_server。在服务器启动和运行之后，controller.py应该作为客户端执行其他python文件，从而连接到它。控制器服务器将永远运行，并创建新的客户端实例，并在必要时向它们发送关机消息。
不幸的是，这是行不通的。没有收到错误，它只是挂起。
controller.py:
async def handleClient(reader, writer):
#handling a connection
addr = writ
浏览 5提问于2020-08-01得票数 0
回答已采纳
2回答
对Async来说还是个新手，所以很抱歉，现在可以非常准确地表达这个问题了。 我尝试按照示例(https://pythonprogramming.net/asyncio-basics-intermediate-python-tutorial/)使用异步，希望获得多个requests.get()的速度优势，而不是以同步的方式进行。 import asyncio
import time
import requests
async def get_text(url):
print(f"Load {url}")
data = requests.get(url).tex
浏览 30提问于2020-04-04得票数 1
1回答
这是我正在使用的示例代码。我想快速解析页面并将结果数据输入到数据库中。然而，在添加一行之后，我的代码运行速度明显变慢了。我知道这与数据库的工作有关。但我不知道该怎么解决它。
如果您对加速此代码有任何其他建议，我将非常感谢您的帮助。
import asyncpg
import asyncio
import aiohttp
from settings import URI, SQL, URLS
class Singleton(type):
_instances = {}
def __call__(cls, *args, **kwargs):
if cls no
浏览 16提问于2020-12-23得票数 0
1回答
在我编写的大多数异步协同器中，我只需要替换函数定义def func() -> async def func()和睡眠time.sleep(s) -> await asyncio.sleep(s)。
可以将标准python函数转换为异步函数，其中所有的time.sleep(s) await asyncio.sleep(s)**?**都转换为。
示例
任务期间的性能
测量任务期间的性能
import asyncio
import random
async def performance_during_task(task):
stop_event = asyncio.Event
浏览 3提问于2022-01-27得票数 1
我试图学习如何在Python中使用异步框架。我有下面的代码，但是它给出了错误：
事件循环在将来完成之前停止。
还请更新在哪里捕获代码的输出。
代码：
import asyncio
import aiopg
dsn = 'dbname=dvdrental user=postgres password=password host=127.0.0.1'
async def go():
pool = await aiopg.create_pool(dsn)
async with pool.acquire() as conn:
async w
浏览 0提问于2018-02-07得票数 0
考虑以下代码，如何在listen()中停止执行？在sock.close()被调用之后，它似乎挂起了。不产生例外情况。
#!/usr/bin/env python3.5
import asyncio, socket
sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
sock.bind(('localhost', 8080))
sock.listen()
sock.setblocking(0)
async def listen():
print('listen')
try:
浏览 8提问于2016-10-11得票数 0
回答已采纳
我和postgres和django合写了我的船坞。尽管如此，当我试图构建时，我收到了下面的错误。我怎么才能修好它？
Is the server running on host "db" (172.22.0.2) and accepting TCP/IP connections on port 5432?
包含错误的日志：
web_1 | Watching for file changes with StatReloader
web_1 | Performing system checks...
web_1 |
web_1 | System check identifie
浏览 4提问于2022-01-07得票数 -1
我成功地导入/安装了Twisted的asyncioreactor并执行了一个简单的异步函数：
from twisted.internet import asyncioreactor
asyncioreactor.install()
from twisted.internet import task
from twisted.internet.defer import inlineCallbacks
from twisted.internet.defer import ensureDeferred
async def sleepy(reactor):
print("SLEEPI
浏览 11提问于2017-02-18得票数 4
回答已采纳
2回答
我想知道在python3.6中并发是如何与异步的工作的。我的理解是，当解释器执行await语句时，它将把它留在那里，直到等待过程完成，然后继续执行其他协同任务。但我在下面的代码中看到的不是这样的。程序同步运行，逐个执行任务。
我的理解和实现代码有什么问题？
import asyncio
import time
async def myWorker(lock, i):
print("Attempting to attain lock {}".format(i))
# acquire lock
with aw
浏览 0提问于2018-03-05得票数 0
回答已采纳
1回答
我正在使用Docker，并且我试图在若干秒后与等待Docker容器完成的另一个任务进行竞争。实际上，我想知道给定的容器是否在我设置的超时内完成。
我有以下代码(改编自)：
container = # ... create container with Docker SDK
timeout = # ... some int
killed = None
# our tasks
async def __timeout():
await asyncio.sleep(timeout)
return True
async def __run():
container.wait()
r
浏览 0提问于2019-01-27得票数 3
回答已采纳
1回答
在我的产品代码中有很多协程，它们在处理请求时被卡在未知的位置。我使用将gdb附加到进程，但它没有显示进程停滞的协程中的确切行，只显示了主堆栈跟踪。下面是一个最小的例子：
import asyncio
async def hello():
await asyncio.sleep(30)
print('hello world')
asyncio.run(hello())
(gdb) py-bt
Traceback (most recent call first):
File "/usr/lib/python3.8/selectors.py",
浏览 7提问于2021-01-13得票数 7
我是PostgreSQL新手，我制作了我的服务器、数据库、表和其他东西，然后回到repl.it discord.py，并尝试运行以下代码：
async def create_db_pool():
client.pg_con = await asyncpg.create_pool(database= "mainbank", user = "postgres", password = password)
client.loop.run_until_complete(create_db_pool())
错误：
Traceback (most recent cal
浏览 1提问于2021-03-04得票数 0
我有一个简单的REST，它使用aiohttp、sqlalchemy和PostgreSQL。
我过去经常在本地运行我的应用程序，在docker容器中运行postgresql。我用localhost:5432连接到它
现在，我正在尝试将我的应用程序添加到docker编写中，并且我在从它连接postgresql时遇到了问题。
这是我的撰写文件：
version: "3"
services:
db:
image: postgres:9.4
container_name: db
restart: always
ports:
- "
浏览 1提问于2019-07-19得票数 2
回答已采纳
最小可重现性示例：
import asyncio
import aiopg
from fastapi import FastAPI, WebSocket
dsn = "dbname=aiopg user=aiopg password=passwd host=127.0.0.1"
app = FastAPI()
class ConnectionManager:
self.count_connections = 0
# other class functions and variables are taken from FastAPI docs
..
浏览 7提问于2022-06-09得票数 1
回答已采纳
问题
在我自己的项目中，我在线程中启动一个事件循环，然后取消任务并安全地关闭另一个线程中的循环。但我失败了。
在阅读了之后，我仍然无法理解如何等待Task.cancel之后真正取消的任务
Python版本:3.7.1
操作系统: windows
下面是我的调试过程。
import threading
import asyncio
import time
async def visit_sth():
print("start sleep")
await asyncio.sleep(3)
print("end sleep")
class
浏览 1提问于2018-12-18得票数 2
2回答
如何在Python中加速异步请求
、、、、
我想从一个网站下载/刮掉5000万条日志记录。我不是一次下载5000万，而是试着用下面的代码一次下载1000万次，但一次只能处理2万次(超过这会引发错误)，所以下载这么多数据变得很费时。目前，以100%|██████████| 20000/20000 [03:48<00:00, 87.41it/s]的速度下载20,000条记录需要3-4分钟，那么如何加快速度呢？
import asyncio
import aiohttp
import time
import tqdm
import nest_asyncio
nest_asyncio.apply()
async def make_n
浏览 12提问于2022-02-23得票数 6
回答已采纳
我有一个不和谐的机器人，我把它托管在heroku上。最近我学习了postgreSQL，它工作得很好，现在当我试图在heroku上托管机器人时，我得到了这个错误： 2020-11-16T08:24:23.418416+00:00 app[worker.1]: Traceback (most recent call last):
2020-11-16T08:24:23.418448+00:00 app[worker.1]: File "Bot.py", line 113, in <module>
2020-11-16T08:24:23.418835+00:00 a
浏览 43提问于2020-11-16得票数 0
2回答
似乎有两种基于发电机的协同线：
从发回：
基于生成器的coroutine：由包装的生成器(def + yield)。如果需要将其视为协同对象，则需要将其包装在types.coroutine中。
简单地说，Python并没有明确称它为“基于生成器的协同器”：
当您编写基于asyncio的Python代码(理想情况下也使用来自asyncio.org的外接程序模块)时，您通常会编写协同函数。包括Python3.4，这类函数是使用第95页“产自(v3-只)”中包含的yield from语句生成的，用@asyncio.coroutine修饰，在第518页中包含“异步协同”；
来自
浏览 5提问于2017-10-19得票数 1
回答已采纳
我正在尝试使用库，但我得到了RuntimeError: Event loop is closed。
这是我的代码和回溯。
代码
import asyncio
import asyncpraw
async def main():
reddit = asyncpraw.Reddit('praw_ini_name')
print('User is', await reddit.user.me())
await reddit.close()
asyncio.run(main())
回溯
User is python_user
Exceptio
浏览 4提问于2021-02-28得票数 3
回答已采纳
1回答
根据，队列有几个实现。我感兴趣的是JoinableQueue和PriorityQueue，因为我希望有一个具有优先级的可连接队列。
看来，我只能在较低版本中获得其中一个功能。也就是说，在3.5中，我可以使用Queue (它是可连接的)和PriorityQueue来区分，但是在3.5以下的python中，有JoinableQueue和PriorityQueue (参见下面的示例)。
是否有可能将它们结合在一起，以获得一个通用的方法，在3.4中获得一个可连接的PriorityQueue？
try:
# Python 3.4.
from asyncio import Joinable
浏览 1提问于2015-07-12得票数 3
回答已采纳
我正在使用套接字和异步实现服务器实现中的生产者-消费者程序。问题是，在封装在ssl连接中的套接字使用时，异步函数sock_recv()似乎不能正常工作。以下是工作代码。
服务器端
import asyncio
import random
import socket
import ssl
SERVER_ADDRESS = (HOST, PORT) = "127.0.0.1", 8881
async def producer(queue, client_connection, event_loop):
while True:
print("Wai
浏览 2提问于2019-06-22得票数 1
回答已采纳
更新:异步只需执行它被告知的操作，您就可以很好地处理这些异常--请参阅我标记为问题解决方案的后续答案。下面是最初的问题，并以稍加修改的例子来澄清这个问题及其解决方案。
我一直在尝试调试一个在很大程度上依赖于异步的库。在编写一些示例代码时，我意识到有时执行键盘中断(CTRL)(很少！)引发了可怕的..。
Task exception was never retrieved
我一直在努力确保我派生的所有任务都能优雅地处理asyncio.CancelledError，在花了太多时间调试它之后，我意识到只有当异步任务之一被阻塞操作卡住时，我才会收到这条错误消息。
封锁？您真的不应该在任务中执行阻塞工作
浏览 5提问于2021-12-17得票数 2
回答已采纳
Python 3.5+支持带有关键字await和IO多路复用库asyncio的coroutinne。我是关于python3.5+是如何用IO多路复用实现协同工作的。
例如，我们有两个函数：
#example-1
async def f1(x, y):
for i in range(10):
r = await x + y
print(r)
async def f2(x, y):
for i in range(10):
await asyncio.sleep(1.0)
r = await f1(x, y)
print(r)
if __nam
浏览 2提问于2020-05-20得票数 0
回答已采纳
当尝试创建具有异步函数和asyncio.run的简单coap服务器时，返回事件循环将关闭。代码如下：
import aiocoap
import aiocoap.resource as resource
class CoAPServer(resource.Resource):
def __init__(self) -> None:
super().__init__()
async def render_get(self, request):
print("Received GET request ...")
浏览 7提问于2021-11-16得票数 1
回答已采纳
我找到了一个异步httpx示例，其中ensure_future工作，但create_task不工作，但我不知道原因。正如我所理解的，create_task是首选的方法，我想知道发生了什么以及如何解决这个问题。
我在上使用了一个异步httpx示例
import asyncio
import httpx
import time
start_time = time.time()
async def get_pokemon(client, url):
resp = await client.get(url)
pokemon = resp.json()
浏览 25提问于2022-10-14得票数 0
回答已采纳
扫码
添加站长 进交流群
领取专属 10元无门槛券
手把手带您无忧上云
相关资讯
热门标签
云服务器
ICP备案
对象存储
实时音视频
即时通信 IM活动推荐
腾讯云开发者
