pip install txpostgres
python
from twisted.internet import defer
from twisted.python import log
from txpostgres import txpostgres
DB_SETTINGS = {
'database': 'mydatabase',
'user': 'myuser',
'password': 'mypassword',
'host': 'localhost',
'port': 5432
}
@defer.inlineCallbacks
def perform_database_operation(connection):
result = yield connection.runQuery("SELECT * FROM mytable")
for row in result:
print(row)
yield connection.runOperation("UPDATE mytable SET column1 = 'new_value' WHERE id = 1")
defer.returnValue(None)
@defer.inlineCallbacks
def connect_to_database():
try:
db_pool = txpostgres.ConnectionPool(None, **DB_SETTINGS)
connection = yield db_pool.start()
log.msg("Connected to the database")
yield perform_database_operation(connection)
yield connection.close()
log.msg("Disconnected from the database")
except Exception as e:
log.err("Error connecting to the database: {error}".format(error=e))
if __name__ == '__main__':
connect_to_database()
