Pysqlx Core 0.2.0 | GitLocker.com Product

Description:

pysqlxcore 0.2.0

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())