Skip to content

Next steps

This section provides some next steps for you to continue learning about PySQLXEngine.

PySQLXEngine has some methods; among them, the query and execute methods are handy.

The methods .query* is used to execute a query that returns data, and .execute is used to run a query that does not return data but only the number of rows affected.

Both methods have the two parameters:

  • sql: The SQL query to be executed.
  • parameters: The parameters to be passed to the query. This parameter is optional.

Note

SQL must be a str with or without named parameters.

Note

Parameters must be a dictionary with the named parameters and their values.

Example of use

For this example, we will use the sqlite database. Create a main.py file and add the code examples below.

Create a table

main.py
 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
from pysqlx_engine import PySQLXEngine

async def main():
    uri = "sqlite:./db.db"
    db = PySQLXEngine(uri=uri)
    await db.connect()

    sql = """
        CREATE TABLE IF NOT EXISTS users (
            id INTEGER PRIMARY KEY, 
            name TEXT, 
            age INTEGER
        );
    """

    resp = await db.execute(sql=sql)

    print("table created: ", resp == 0)

import asyncio
asyncio.run(main())
main.py
 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
from pysqlx_engine import PySQLXEngineSync

def main():
    uri = "sqlite:./db.db"
    db = PySQLXEngineSync(uri=uri)
    db.connect()

    sql = """
        CREATE TABLE IF NOT EXISTS users (
            id INTEGER PRIMARY KEY, 
            name TEXT, 
            age INTEGER
        );
    """

    resp = db.execute(sql=sql)

    print("table created: ", resp == 0)

# running the code
main()

Running the code using the terminal

$ python3 main.py

table created: True

Insert data

main.py
 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
from pysqlx_engine import PySQLXEngine

async def main():
    uri = "sqlite:./db.db"
    db = PySQLXEngine(uri=uri)
    await db.connect()

    sql = "INSERT INTO users (name, age) VALUES (:name, :age)"
    params = {"name": "John", "age": 20}
    resp = await db.execute(sql=sql, parameters=params)

    print("rows affected: ", resp)

import asyncio
asyncio.run(main())
main.py
 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
from pysqlx_engine import PySQLXEngineSync

def main():
    uri = "sqlite:./db.db"
    db = PySQLXEngineSync(uri=uri)
    db.connect()

    sql = "INSERT INTO users (name, age) VALUES (:name, :age)"
    params = {"name": "John", "age": 20}
    resp = db.execute(sql=sql, parameters=params)

    print("rows affected: ", resp)

# running the code
main()

Running the code using the terminal

$ python3 main.py

rows affected: 1

Select data

main.py
 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
from pysqlx_engine import PySQLXEngine

async def main():
    uri = "sqlite:./db.db"
    db = PySQLXEngine(uri=uri)
    await db.connect()

    sql = "SELECT * FROM users"
    resp = await db.query(sql=sql)

    print("rows selected: ", resp)

import asyncio
asyncio.run(main())
main.py
 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
from pysqlx_engine import PySQLXEngineSync

def main():
    uri = "sqlite:./db.db"
    db = PySQLXEngineSync(uri=uri)
    db.connect()

    sql = "SELECT * FROM users"
    resp = db.query(sql=sql)

    print("rows selected: ", resp)

# running the code
main()

Running the code using the terminal

$ python3 main.py

rows selected: [BaseRow(id=1, name='John', age=20)]

Update data

main.py
 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
from pysqlx_engine import PySQLXEngine

async def main():
    uri = "sqlite:./db.db"
    db = PySQLXEngine(uri=uri)
    await db.connect()

    sql = "UPDATE users SET age = :age WHERE id = :id"
    params = {"age": 21, "id": 1}
    resp = await db.execute(sql=sql, parameters=params)

    print("rows affected: ", resp)

import asyncio
asyncio.run(main())
main.py
 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
from pysqlx_engine import PySQLXEngineSync

def main():
    uri = "sqlite:./db.db"
    db = PySQLXEngineSync(uri=uri)
    db.connect()

    sql = "UPDATE users SET age = :age WHERE id = :id"
    params = {"age": 21, "id": 1}
    resp = db.execute(sql=sql, parameters=params)

    print("rows affected: ", resp)

# running the code
main()

Running the code using the terminal

$ python3 main.py

rows affected: 1

Delete data

main.py
 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
from pysqlx_engine import PySQLXEngine

async def main():
    uri = "sqlite:./db.db"
    db = PySQLXEngine(uri=uri)
    await db.connect()

    sql = "DELETE FROM users WHERE id = :id"
    params = {"id": 1}
    resp = await db.execute(sql=sql, parameters=params)

    print("rows affected: ", resp)

import asyncio
asyncio.run(main())
main.py
 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
from pysqlx_engine import PySQLXEngineSync

def main():
    uri = "sqlite:./db.db"
    db = PySQLXEngineSync(uri=uri)
    db.connect()

    sql = "DELETE FROM users WHERE id = :id"
    params = {"id": 1}
    resp = db.execute(sql=sql, parameters=params)

    print("rows affected: ", resp)

# running the code
main()

Running the code using the terminal

$ python3 main.py

rows affected: 1

Drop table

main.py
 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
from pysqlx_engine import PySQLXEngine

async def main():
    uri = "sqlite:./db.db"
    db = PySQLXEngine(uri=uri)
    await db.connect()

    sql = "DROP TABLE users"
    resp = await db.execute(sql=sql)

    print("table dropped: ", resp == 0)

import asyncio
asyncio.run(main())
main.py
 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
from pysqlx_engine import PySQLXEngineSync

def main():
    uri = "sqlite:./db.db"
    db = PySQLXEngineSync(uri=uri)
    db.connect()

    sql = "DROP TABLE users"
    resp = db.execute(sql=sql)

    print("table dropped: ", resp == 0)

# running the code
main()

Running the code using the terminal

$ python3 main.py

table dropped: True