asyncpg_postgres_connect_query_with_type_hints.py

1import asyncio
2import asyncpg
3
4async def run():
5    # Establish a connection to an existing database named "test"
6    # as a "postgres" user.
7    conn = await asyncpg.connect(user='postgres', password='password',
8                                 database='test', host='127.0.0.1')
9    
10    # Execute a statement to create a new table.
11    await conn.execute('''
12        CREATE TABLE IF NOT EXISTS users(
13            id serial PRIMARY KEY,
14            name text,
15            dob date
16        )
17    ''')
18
19    # Insert a record into the created table.
20    await conn.execute('''
21        INSERT INTO users(name, dob) VALUES($1, $2)
22    ''', 'Bob', datetime.date(1984, 3, 1))
23
24    # Select a row from the table.
25    row = await conn.fetchrow(
26        'SELECT * FROM users WHERE name = $1', 'Bob')
27    
28    # With asyncpg-stubs installed, IDEs/Mypy will correctly 
29    # identify 'row' as an asyncpg.Record
30    print(row)
31
32    # Close the connection.
33    await conn.close()
34
35if __name__ == "__main__":
36    import datetime
37    asyncio.run(run())