upd, add sqlite

This commit is contained in:
ge 2023-07-25 01:40:12 +03:00
parent a511550f71
commit ff69b68ba3
5 changed files with 51 additions and 40 deletions

2
.gitignore vendored
View File

@ -1,3 +1,5 @@
__pychache__/
*.pyc
dist/
db.sqlite3
*.db

View File

@ -2,17 +2,17 @@
**Quicker** is a pythonic 🐍 tool for querying databases.
Quicker wraps popular Python libraries:
Quicker wraps Python libraries:
- `mysqlclient` for MySQL.
- `psycopg2` for PostgreSQL.
- `sqlite` from Python standard library for SQLite (not implemented yet).
- `sqlite3` from Python standard library for SQLite.
# Why is it needed?
Briefly — just make queries without bothering to learn the abstractions of the library.
At work, I periodically have to make queries to different databases and then somehow process the information received. This may be necessary for one-time use, so I don't want to write a lot of code. You may also want to do all the work right in the interactive Python shell.
At work, I periodically have to make queries to different databases and then somehow process the information. I am not an analyst and Quicker is not intended for analysts. The main use of the library is to quickly execute raw SQL queries into the database. You may want to do this entirely in the Python interactive shell. Quicker interface is as simple as possible, thanks to which lazy system administrators can now effortlessly extract data from the database.
Quicker interface is as simple as possible, thanks to which lazy system administrators can now effortlessly extract data from the database.
Of course, this library **should not be used in production**. This is just a small assistant in routine tasks.
@ -135,7 +135,7 @@ logging.basicConfig(level=logging.DEBUG)
## Direct access to Cursor object
```python
from quicker import Connection, make_list
from quicker import Connection, mklist
# config declaration here...
@ -144,5 +144,5 @@ with Connection(**config) as db:
users = db.cursor.fetchall()
# Note: "users" is tuple! Convert it to list of dicts!
colnames = [desc[0] for desc in db.cursor.description]
users_list = make_list(colnames, users)
users_list = mklist(colnames, users)
```

View File

@ -1,6 +1,6 @@
[tool.poetry]
name = "quicker"
version = "0.2.0"
version = "0.3.0"
description = "Query databases quickly."
authors = ["ge <ge@nixhacks.net>"]
license = "Unlicense"

View File

@ -1,3 +1,13 @@
#
# .88888. oo dP
# d8' `8b 88
# 88 88 dP dP dP .d8888b. 88 .dP .d8888b. 88d888b.
# 88 db 88 88 88 88 88' `"" 88888" 88ooood8 88' `88
# Y8. Y88P 88. .88 88 88. ... 88 `8b. 88. ... 88
# `8888PY8b `88888P' dP `88888P' dP `YP `88888P' dP
#
# Quicker -- pythonic tool for querying databases.
import logging
import importlib.util
from enum import Enum
@ -21,10 +31,12 @@ def _import(module_name: str, symbol: Optional[str] = None):
return module
def make_list(
column_names: List[str], rows: Union[tuple, Tuple[dict, ...]]
) -> List[dict]:
"""Convert output to list of dicts from tuples."""
def mklist(column_names: List[str], rows: Union[tuple, Tuple[dict, ...]]) -> List[dict]:
"""
Convert output to list of dicts from tuples. `rows` can be
default tuple or tuple of dicts if MySQL provider is used with
MySQLdb.cursors.DictCursor cursor class.
"""
data = []
for row in rows:
if isinstance(row, dict):
@ -38,7 +50,6 @@ def make_list(
class Provider(str, Enum):
MYSQL = 'mysql'
POSTGRES = 'postgres'
SQLITE = 'sqlite'
@ -55,26 +66,17 @@ class Connection:
>>>
"""
def __init__(self,
provider: Provider = None,
commit: bool = True,
**kwargs
):
if not provider:
raise ValueError('Database provider is not set')
def __init__(self, provider: Provider, commit: bool = True, **kwargs):
self._provider = Provider(provider)
self._commit = commit
self._connection_args = kwargs
def __enter__(self):
logger.debug(f'Database provider={self._provider}')
# -- MySQL / MariaDB --
if self._provider == Provider.MYSQL:
MySQLdb = _import('MySQLdb')
DictCursor = _import('MySQLdb.cursors', 'DictCursor')
try:
if self._connection_args['cursorclass']:
cursorclass = DictCursor
cursorclass = self._connection_args.pop('cursorclass')
except KeyError:
cursorclass = DictCursor
self._connection = MySQLdb.connect(
@ -83,23 +85,35 @@ class Connection:
)
logger.debug('Session started')
self._cursor = self._connection.cursor()
return Query(self)
# -- PostgreSQL --
self._queryobj = Query(self)
if self._provider == Provider.POSTGRES:
psycopg2 = _import('psycopg2')
dbname = self._connection_args.pop('database')
self._connection_args['dbname'] = dbname
self._connection = psycopg2.connect(**self._connection_args)
self._cursor = self._connection.cursor()
return Query(self)
self._queryobj = Query(self)
if self._provider == Provider.SQLITE:
sqlite3 = _import('sqlite3') # Python may built without SQLite
self._connection = sqlite3.connect(**self._connection_args)
self._cursor = self._connection.cursor()
self._queryobj = Query(self)
def __enter__(self):
return self._queryobj
def __exit__(self, exception_type, exception_value, exception_traceback):
self.close()
def close(self):
if self._commit:
logger.debug('Commiting changes into database')
self._connection.commit()
logger.debug('Changes commited into database')
logger.debug('Closing cursor and connection')
self._cursor.close()
self._connection.close()
logger.debug('Connection closed')
class Query:
@ -128,13 +142,17 @@ class Query:
self._fetchall = self._cursor.fetchall()
except pgProgrammingError as e:
self._fetchall = None
if self._provider == Provider.SQLITE:
self._cursor.execute(*args, **kwargs)
logger.debug(f'sqlite3 ran: {args}')
self._fetchall = self._cursor.fetchall()
if self._fetchall is not None:
self._colnames = []
if self._cursor.description is not None:
self._colnames = [
desc[0] for desc in self._cursor.description
]
return make_list(self._colnames, self._fetchall)
return mklist(self._colnames, self._fetchall)
return None
def commit(self) -> None:

View File

@ -1,9 +0,0 @@
# ____ _ __
# / __ \__ __(_)____/ /_____ _____
# / / / / / / / / ___/ //_/ _ \/ ___/
# / /_/ / /_/ / / /__/ ,< / __/ /
# \___\_\__,_/_/\___/_/|_|\___/_/
#
# Quicker -- pythonic tool for querying databases
from .main import Connection, make_list