Using psycopg2 with PostgreSQL

I had been using MySql my whole life until recently I got my hands dirty on PostgreSQL in one of projects. I must tell, switching to PostgreSQL has been very easy. It has got some very cool and robust features. Let’s not talk about that here. When using python, psycopg2 is one of the mostly used database adapter. It is fairly stable and got a good community support. We used aiopg, which is a library for accessing a PostgreSQL database with asyncio. In this post, I will try to mention few important things which I came across.

1. DictCursor:

dict_cur = conn.cursor(cursor_factory=psycopg2.extras.DictCursor)

helps in fetching data from the database as a Python dictionary where we can easily get columns against their names. A plain Curson gives values against their index which can be sometimes painful. Say we have to fetch a row for id = 3 from user table and we have to use couple of fields as: name, age and gender and we do not want to use 10 other fields. Using dictcursor we can get these data as :

row.get('name'), row.get('age') and row.get('gender')

against :

row[2], row.get[4] and row.get[10], where 2, 4, and 10 are the orders of the required field

Some code:

cur = conn.cursor(cursor_factory=psycopg2.extras.DictCursor)
query = """SELECT * FROM {} where user_id = %s""".format(DBOperations.TABLE_NAME)
yield from cur.execute(query, (10, ))
row = yield from cur.fetchall()
return row

2. Single insert for multiple rows:

We might want to execute multiple insert in one query.

insert into user address ('name1', 'address1'), ('name1', 'address2'), ('name1', 'address3')

we have to construct the string and execute the query, which can be done as below:

def set_address(self, user_id, address_ids:list):
	tup = [(user_id, aid) for aid in address_ids]
	args_str = ",".join([str(s) for s in tup])
	insert = yield from cur.execute("INSERT INTO user_address VALUES " + args_str)

3. Searching in a jsonb array:

One of the cool datatypes in PostgreSql is jsonb array. PS has made sure that querying this array is easy. Sometime we may need to search for a particular key in the jsons, say a user has got many addresses in various cities and we need to look for all the users who have address in Mumbai.

def find_user_address_by_city(cls, cur, city: str):
        array_str = "%s"
        query = """SELECT * FROM user WHERE to_json(array(SELECT jsonb_array_elements(address) ->> 'city'))::jsonb ?|
        yield from cur.execute(query, tuple(city))
        rows = yield from cur.fetchall()
        return rows

I will try to add other things as and when I get them.