pip install tornpsql
- Query via
args
orkwargs
- Support Search Path
- Query from Files
- Pubsub
- Retrieve notices (
raise notice 'something';
) vialist(db.notices)
# Method 1
import tornpsql
db = tornpsql.Connection("postgres://postgres-user:[email protected]:5432/postgres-db")
import tornpsql
db = tornpsql.Connection("127.0.0.1", "database", "postgres", "postgres-user", 5432)
# get one
db.get("SELECT col from table where col = %s limit 1", value)
# >>> {"col": "value"}
# get many
db.query("SELECT col from table where col = %s limit 2", value)
# >>> [{"col": "value"}, {"col": "value"}]
Set the search_path
for the duration of the proceeding query.
# set the "default" search_path to public
db = tornpsql.Connection(search_path="public")
results = db.path("another_schema").query("select column from mytable")
results = db.path("another_schema,and_another").query("select column from mytable")
# this will use search path: "public"
results = db.query("select column from mytable")
-- main.sql
create table example (id serial primary key);
\ir other.sql
-- other.sql
insert into example values (1, 2, 3);
db = tornpsql.Connection()
db.file("main.sql")
db = tornpsql.Connection()
pubsub = db.pubsub()
pubsub.subscribe( ("channel_1", ) )
for notify in pubsub.listen():
print notify.pid, notify.channel, notify.payload
class PubSubThread(threading.Thread):
def __init__(self):
threading.Thread.__init__(self)
self.db = tornpsql.Connection()
self.pubsub = self.db.pubsub()
self.pubsub.subscribe( ("channel_1", "channel_2") )
def run(self):
for notify in self.pubsub.listen():
print notify.pid, notify.channel, notify.payload
pubsub = PubSubThread()