This commit is contained in:
ge 2023-07-09 03:47:50 +03:00
commit c452f7519c
7 changed files with 234 additions and 0 deletions

3
.gitignore vendored Normal file
View File

@ -0,0 +1,3 @@
__pychache__/
*.pyc
dist/

86
README.md Normal file
View File

@ -0,0 +1,86 @@
Quicker is a pythonic tool for querying databases.
Quicker wraps popular Python packages:
- `mysqlclient` for MySQL.
- `psycopg2` for PostgreSQL.
- Python builtin `sqlite` for SQLite.
Connection parameters will passed to "backend" module as is.
# `Connection` class
`Connection` is context manages and must be used with `with` keyword. `Connection` returns `Query` callable object. `Query` can be called in this ways:
```python
with Connection(**config) as db:
db.exec("sql query here...")
db.query(sql query here...") # query is alias for exec()
# Query is callable and you can also do this:
with Connection(**config) as query:
query("sql query here...")
```
# `Query`
`Query` cannot be called itself, you must use `Connection` to correctly initialise `Query` object.
Methods and properties:
- `query()`, `exec()`. Execute SQL. There is You can use here this syntax: `query('SELECT * FROM users WHERE id = %s', (15,))`.
- `commit()`. Write changes into database.
- `cursor`. Call [MySQLdb Cursor object](https://mysqlclient.readthedocs.io/user_guide.html#cursor-objects) methods directly.
# Examples
## SELECT
```python
import json
from quicker import Connection
config = {
'provider': 'mysql',
'host': '127.0.0.1',
'port': 3306,
'user': 'myuser',
'database': 'mydb',
'password': 'example',
}
with Connection(**config) as query:
users = query("SELECT * FROM `users`")
print(json.dumps(users, indent=4))
```
`users` will presented as list of dicts:
```python
[
{
'id': 1,
'name': 'test',
'email': 'noreply@localhost'
},
{
'id': 2,
'name': 'user1',
'email': 'my@example.com'
}
]
```
## Change database
```
from quicker import Connection
with Connection(provider='mysql', read_default_file='~/.my.cnf') as db:
db.query("INSERT INTO users VALUE (3, 'user2', 'user2@example.org')")
```
Quicker by default make commit after closing context. Set option `commit=False` to disable automatic commit.

21
docker-compose.yaml Normal file
View File

@ -0,0 +1,21 @@
version: '3.1'
services:
db:
image: mysql:8
command: --default-authentication-plugin=mysql_native_password
restart: always
environment:
MYSQL_USER: myuser
MYSQL_PASSWORD: example
MYSQL_DATABASE: mydb
MYSQL_ROOT_PASSWORD: example
ports:
- 3306:3306
adminer:
image: adminer
restart: always
ports:
- 8080:8080

23
poetry.lock generated Normal file
View File

@ -0,0 +1,23 @@
# This file is automatically @generated by Poetry 1.4.2 and should not be changed by hand.
[[package]]
name = "mysqlclient"
version = "2.2.0"
description = "Python interface to MySQL"
category = "main"
optional = false
python-versions = ">=3.8"
files = [
{file = "mysqlclient-2.2.0-cp310-cp310-win_amd64.whl", hash = "sha256:68837b6bb23170acffb43ae411e47533a560b6360c06dac39aa55700972c93b2"},
{file = "mysqlclient-2.2.0-cp311-cp311-win_amd64.whl", hash = "sha256:5670679ff1be1cc3fef0fa81bf39f0cd70605ba121141050f02743eb878ac114"},
{file = "mysqlclient-2.2.0-cp38-cp38-win_amd64.whl", hash = "sha256:004fe1d30d2c2ff8072f8ea513bcec235fd9b896f70dad369461d0ad7e570e98"},
{file = "mysqlclient-2.2.0-cp39-cp39-win_amd64.whl", hash = "sha256:9c6b142836c7dba4f723bf9c93cc46b6e5081d65b2af807f400dda9eb85a16d0"},
{file = "mysqlclient-2.2.0-pp38-pypy38_pp73-win_amd64.whl", hash = "sha256:955dba905a7443ce4788c63fdb9f8d688316260cf60b20ff51ac3b1c77616ede"},
{file = "mysqlclient-2.2.0-pp39-pypy39_pp73-win_amd64.whl", hash = "sha256:530ece9995a36cadb6211b9787f0c9e05cdab6702549bdb4236af5e9b535ed6a"},
{file = "mysqlclient-2.2.0.tar.gz", hash = "sha256:04368445f9c487d8abb7a878e3d23e923e6072c04a6c320f9e0dc8a82efba14e"},
]
[metadata]
lock-version = "2.0"
python-versions = "^3.11"
content-hash = "462894cf2fdfd3121ca0c3a328dce7012ba1cecd25ba04d8a853c1fc21dde8a3"

15
pyproject.toml Normal file
View File

@ -0,0 +1,15 @@
[tool.poetry]
name = "quicker"
version = "0.1.0"
description = "Query databases quickly."
authors = ["ge <ge@nixhacks.net>"]
license = "Unlicense"
readme = "README.md"
[tool.poetry.dependencies]
python = "^3.11"
mysqlclient = "^2.2.0"
[build-system]
requires = ["poetry-core"]
build-backend = "poetry.core.masonry.api"

1
quicker/__init__.py Normal file
View File

@ -0,0 +1 @@
from .main import Connection

85
quicker/main.py Normal file
View File

@ -0,0 +1,85 @@
import importlib.util
from enum import Enum
class Provider(str, Enum):
MYSQL = 'mysql'
POSTGRES = 'postgres'
SQLITE = 'sqlite'
class Connection:
"""
Connection is context manager that allows to establish connection
with database.
"""
def __init__(self,
provider: Provider = None,
commit: bool = True,
**kwargs
):
if not provider:
raise ValueError('Database provider is not set')
self._provider = provider
self._commit = commit
self._connection_args = kwargs
def __enter__(self):
if self._provider == Provider.MYSQL:
MySQLdb = self._import('MySQLdb')
self._connection = MySQLdb.connect(**self._connection_args)
self._cursor = self._connection.cursor()
return Query(self)
def __exit__(self, exception_type, exception_value, exception_traceback):
if self._provider == Provider.MYSQL:
if self._commit:
self._connection.commit()
self._cursor.close()
self._connection.close()
def _import(self, lib: str):
spec = importlib.util.find_spec(lib)
if spec is None:
raise ImportError(f"Module '{lib}' not found.")
module = importlib.util.module_from_spec(spec)
spec.loader.exec_module(module)
return module
class Query:
def __init__(self, connect):
self._provider = connect._provider
self._connection = connect._connection
self.cursor = self._connection.cursor()
def __call__(self, *args, **kwargs):
return self.exec(*args, **kwargs)
def query(self, *args, **kwargs):
return self.exec(*args, **kwargs)
def exec(self, *args, **kwargs):
"""Execute SQL query and return output if available."""
self.cursor.execute(*args, **kwargs)
self._fetchall = self.cursor.fetchall()
if self._fetchall is not None:
if self.cursor.description is not None:
self._field_names = [i[0] for i in self.cursor.description]
return self._conv(self._field_names, self._fetchall)
def commit(self):
"""Commit changes into database."""
if self._provider == Provider.MYSQL:
self._connection.commit()
def _conv(self, field_names, rows) -> list[dict]:
data = []
for row in rows:
item = {}
for i in range(len(row)):
item[field_names[i]] = row[i]
data.append(item)
return data