from twisted.internet import defer
from twisted.enterprise import adbapi
dbpool = adbapi.ConnectionPool("psycopg2", user='username', host='localhost', database='mydb')
@defer.inlineCallbacks
def query_data():
conn = yield dbpool.connection()
try:
result = yield conn.runQuery("SELECT * FROM mytable")
print(result)
finally:
conn.close()
query_data()
from twisted.internet import defer
from txpostgres import txpostgres
@defer.inlineCallbacks
def query_data():
conn = yield txpostgres.Connection().connect(host="localhost", user="username", password="password", database="mydb")
try:
result = yield conn.runQuery("SELECT * FROM mytable")
print(result)
finally:
conn.disconnect()
query_data()
from twisted.internet import defer
from txpostgres import txpostgres
@defer.inlineCallbacks
def query_data(name):
conn = yield txpostgres.Connection().connect(host="localhost", user="username", password="password", database="mydb")
try:
result = yield conn.runQuery("SELECT * FROM mytable WHERE name = %s", (name,))
print(result)
finally:
conn.disconnect()
query_data("John")
