shell
pip install txpostgres
python
from twisted.internet import defer, reactor
from twisted.python import log
from twisted.enterprise import adbapi
import txpostgres
@defer.inlineCallbacks
def connect_to_postgresql():
dbpool = adbapi.ConnectionPool(
"txpostgres",
host='localhost',
database='mydatabase',
user='myuser',
password='mypassword'
)
try:
conn = yield dbpool.connection()
result = yield conn.runQuery("SELECT * FROM mytable")
for row in result:
print(row)
except Exception as e:
print("Error:", e)
finally:
conn.close()
dbpool.close()
yield dbpool.wait()
if __name__ == '__main__':
connect_to_postgresql().addBoth(lambda _: reactor.stop())
reactor.run()
