Skip to content

PySQLXEngine

PySQLXEngine Logo

PySQLXEngine, a fast and minimalist SQL Engine

CI Coverage Package version Supported Python versions Downloads

Documentation: https://carlos-rian.github.io/pysqlx-engine/

Source Code: https://github.com/carlos-rian/pysqlx-engine

PySQLXEngine supports the option of sending Raw SQL to your database.

The PySQLXEngine is a minimalist SQL Engine.

The PySQLXEngine was created and thought to be minimalistic, but very efficient. The core is write in Rust, making communication between Databases and Python more efficient.

All SQL executed using PySQLXEngine is atomic; only one instruction is executed at a time. Only the first one will be completed if you send an Insert and a select. This is one of the ways to handle SQL ingestion. As of version 0.2.0, PySQLXEngine supports transactions, where you can control BEGIN, COMMIT, ROLLBACK , ISOLATION LEVEL, etc. as you wish.

Note

Minimalism is not the lack of something, but having exactly what you need.

PySQLXEngine aims to expose an easy interface for you to communicate with the database in a simple, intuitive way and with good help through documentation, autocompletion, typing, and good practices.

Database Support:

OS Support:

Installation

UV

$ uv add pysqlx-engine

---> 100%

PIP

$ pip install pysqlx-engine

---> 100%

Poetry

$ poetry add pysqlx-engine

---> 100%

Running

Create a main.py file and add the code examples below.

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

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

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

    sql = "INSERT INTO users (name, age) VALUES (:n, :a);"
    await db.execute(sql=sql, parameters={"n": "Rian", "a": 28})
    await db.execute(sql=sql, parameters={"n": "Mary", "a": 25})

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

    print(rows)

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
22
23
24
from pysqlx_engine import PySQLXEngineSync

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

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

    sql = "INSERT INTO users (name, age) VALUES (:n, :a);"
    db.execute(sql=sql, parameters={"n": "Rian", "a": 28})
    db.execute(sql=sql, parameters={"n": "Mary", "a": 25})

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

    print(rows)

# runnig the code
main()

Running the code using the terminal

$ python3 main.py

[BaseRow(id=1, name='Rian', age=28),  BaseRow(id=2, name='Carlos', age=29)]