pysqlx-core

pysqlx-core

pysqlx-core is an extremely fast Python library for communicating with various SQL databases.

This package provides the core functionality for PySQLX-Engine.

The package is currently a work in progress and subject to significant change.

pysqlx-core will be a separate package, required by pysqlx-engine.

This package is written entirely in Rust and compiled as a Python library using PyO3 and PyO3-Asyncio.

This core is not so friendly, but maybe you want to use it, feel free to suggest improvements.

Supported databases

• SQLite
• PostgreSQL
• MySQL
• Microsoft SQL Server

Supported Python versions

• Python >= 3.8

Supported operating systems

• Linux
• MacOS
• Windows

Example of installation:

PIP

$ pip install pysqlx-core

Poetry

$ poetry add pysqlx-core

Example of usage:

import pysqlx_core
import asyncio

async def main(sql):
    # Create a connection 
    db = await pysqlx_core.new(uri="postgresql://postgres:postgrespw@localhost:49153")
    
    # Create a table
    stmt = pysqlx_core.PySQLxStatement(
        provider="postgresql", 
        sql="""
            CREATE TABLE IF NOT EXISTS test (
                id SERIAL PRIMARY KEY,
                name VARCHAR(255) NOT NULL
            );
        """)
    await db.execute(stmt=stmt)

    # Insert a row and return quantity rows affected
    insert = pysqlx_core.PySQLxStatement(
        provider="postgresql", 
        sql="INSERT INTO test (name) VALUES (:name);",
        params={"name": "Carlos"}
    )
    await db.execute(stmt=insert)

    # can you see the sql and params pre builded
    print("SQL:", insert.sql())
    # output: INSERT INTO test (name) VALUES ($1);
    print("PARAMS:", insert.params())
    # output: ['Carlos']

    # Select all rows, return a class PySQLxResponse
    result = await db.query_typed(stmt=pysqlx_core.PySQLxStatement(
            provider="postgresql", 
            sql="SELECT * FROM test;"
        )
    )
    # get first row
    row = result.get_first() # Dict[str, Any] 
    # get all rows
    rows = result.get_all() # List[Dict[str, Any]]
    # return the db 'types' to Pydantic BaseModel
    types = result.get_types() # Dict[str, str] 

    # Select all rows, return how List[Dict[str, Any]]
    rows = await db.query_all(pysqlx_core.PySQLxStatement(provider="postgresql", sql="SELECT * FROM test;"))

    # close? no need 👌-> auto-close when finished programmer or go out of context..
    
asyncio.run(main())