added special filters #1

.pre-commit-config.yaml

@@ -5,14 +5,12 @@ repos:
  - id: check-merge-conflict
  - id: debug-statements
  - id: detect-private-key
- - repo: https://github.com/pre-commit/mirrors-isort
- rev: 'v4.3.21'
- hooks:
- - id: isort
  - repo: https://github.com/psf/black
  rev: 19.3b0
  hooks:
  - id: black
+ args:
+ - minislite
  - repo: local
  hooks:
  - id: mypy
@@ -30,3 +28,8 @@ repos:
  entry: pylint minislite
  language: system
  pass_filenames: false
+ - id: isort
+ name: isort
+ entry: isort -rc .
+ language: system
+ pass_filenames: false

README.md

@@ -5,6 +5,7 @@ MiniSLite is a secure and mini SQLite ORM module.
 
 ### From Pypi
 
+The script is [available on PyPI](https://pypi.org/project/minislite/). To install with pip:
 ```
 pip install minislite
 ```
@@ -51,13 +52,16 @@ database.add_model(Person)
 ### Create and Update Object
 
 ```python
-person1 = Person(name="mini", last_name="slite")
+person1 = Person(name="mini", last_name="slite") # not created
 person1.age = 1
-person1.save()
+person1.save() # created
 
-person2 = Person.objects.create(name="mini2", last_name="slite")
+person2 = Person.objects.create(name="mini2", last_name="slite") # created
 person2.age = 2
-person2.save()
+person2.save() # updated
+
+person3 = Person.objects.create(name="mini3", last_name="slite", age=10) # created
+person4 = Person.objects.create(name="mini4", last_name="slite", age=20) # created
 ```
 
 ### Update All Objects
@@ -82,6 +86,16 @@ person_obj = Person.objects.get(name="mini")
 person_obj.delete()
 ```
 
+**Special Filters**
+* ``gt``, ``gte``, ``lt``, ``lte`` for integer, bool and float fields
+* ``contains``, ``startswith``, ``endswith`` for string fields
+
+```python
+older_than_10 = Person.objects.filter(age__gt=10)[0]
+print(older_than_10.name) # mini4
+```
+
+
 ### Delete All Objects
 
 ```python
@@ -98,14 +112,15 @@ database.drop_model(Person)
 ### Exceptions
 
 ```python
-from minislite.exceptions import RecordNotFoundError, AlreadyExistsError, DatabaseNotFound, \
- AreYouSure
+from minislite.exceptions import RecordNotFoundError, AlreadyExistsError, DatabaseNotFoundError, \
+ AreYouSureError, WhereOperatorError
 ```
 
 * **RecordNotFoundError:** If you use ``objects.get()`` and that is not found in database
 * **AlreadyExistsError:** Raise this exception when an object creating or saving. Check your ``unique=True`` fields and ``unique_together`` fields.
-* **DatabaseNotFound:** Cannot use TableManager if you don't initialize ``MiniSLiteDb()``
-* **AreYouSure:** Raise this exception if you want to delete all objects(``objects.delete()``) in model. Add ``i_am_sure=True`` arguments.
+* **DatabaseNotFoundError:** Cannot use TableManager if you don't initialize ``MiniSLiteDb()``
+* **AreYouSureError:** Raise this exception if you want to delete all objects(``objects.delete()``) in model. Add ``i_am_sure=True`` arguments.
+* **WhereOperatorError:** Raise this exception if you use integer operator for string field. Or vice versa.
 
 ## Development and Contribution
 See; [CONTRIBUTING.md](CONTRIBUTING.md)

minislite/exceptions.py

@@ -6,9 +6,13 @@ class AlreadyExistsError(ValueError):
  pass
 
 
-class DatabaseNotFound(ConnectionError):
+class DatabaseNotFoundError(ConnectionError):
  pass
 
 
-class AreYouSure(ValueError):
+class AreYouSureError(ValueError):
+ pass
+
+
+class WhereOperatorError(ValueError):
  pass

minislite/models.py

@@ -16,7 +16,7 @@ class MiniSLiteModel:
 
  def __init__(self, **kwargs):
  self.reload_attributes(**kwargs)
- self.manager = TableManager(table_name=self.get_table_name())
+ self.manager = TableManager(model=self)
 
  def reload_attributes(self, **kwargs) -> None:
  field_list = self.get_fields()

minislite/objects.py

@@ -3,21 +3,19 @@
 
 # First Party
 from minislite.tables import TableManager
-from minislite.exceptions import AreYouSure, AlreadyExistsError, RecordNotFoundError
+from minislite.exceptions import AreYouSureError, AlreadyExistsError, RecordNotFoundError
 
 
 class Objects:
- table_name = None
  unique_together = None
 
  klass: Any
  manager: Any
 
  def __get__(self, instance, owner):
- self.table_name = owner.get_table_name()
  self.unique_together = owner.get_unique_together()
  self.klass = owner
- self.manager = TableManager(table_name=self.table_name)
+ self.manager = TableManager(model=owner)
 
  return self
 
@@ -78,6 +76,6 @@ def create(self, *args, **kwargs):
 
  def delete(self, i_am_sure=None, object_id=None) -> None:
  if object_id is None and i_am_sure is None:
- raise AreYouSure("Cleaning all data in tables. Are you sure? Use 'i_am_sure=True'")
+ raise AreYouSureError("Cleaning all data in tables. Are you sure? Use 'i_am_sure=True'")
 
  self.manager.delete(object_id)

minislite/tables.py

@@ -1,33 +1,75 @@
 # Standard Library
 import os
 import sqlite3
+from typing import Any, Dict, Tuple, Optional
 
 # First Party
-from minislite.exceptions import DatabaseNotFound
+from minislite.exceptions import WhereOperatorError, DatabaseNotFoundError
 
 
 class TableManager:
- def __init__(self, table_name) -> None:
+ def __init__(self, model) -> None:
  db_file = os.environ.get("MINISLITE_DB_PATH", None)
  if db_file is None:
- raise DatabaseNotFound("Database not defined.")
+ raise DatabaseNotFoundError("Database not defined.")
 
- self.table_name = table_name
+ self.model = model
+ self.table_name = model.get_table_name()
  self.connection = sqlite3.connect(db_file)
  self.connection.row_factory = sqlite3.Row
 
  self.cursor = self.connection.cursor()
 
+ self.integer_operators = {
+ "gt": {"operator": ">", "value_format": "{0}"},
+ "gte": {"operator": ">=", "value_format": "{0}"},
+ "lt": {"operator": "<=", "value_format": "{0}"},
+ "lte": {"operator": "<=", "value_format": "{0}"},
+ }
+ self.string_operators = {
+ "contains": {"operator": "LIKE", "value_format": "%{0}%"},
+ "startswith": {"operator": "LIKE", "value_format": "{0}%"},
+ "endswith": {"operator": "LIKE", "value_format": "%{0}"},
+ }
+
+ self.operators = {
+ int: self.integer_operators,
+ bool: self.integer_operators,
+ float: self.integer_operators,
+ str: self.string_operators,
+ }
+
+ def get_special_key_values(self, key, value) -> Tuple[Any, Optional[str], Any]:
+ if value is None:
+ return key, "IS", None
+
+ if "__" in key:
+ operator_key = key.split("__")[1]
+ key = key.split("__")[0]
+ field_type = getattr(self.model, key).field_type
+
+ operator_list: Optional[Dict[str, Dict[str, str]]]
+ operator_list = self.operators.get(field_type)
+ parameters = operator_list.get(operator_key, None) # type: ignore
+ if parameters is None:
+ raise WhereOperatorError(f"{operator_key} operator not found for {key}")
+
+ operator = parameters.get("operator")
+ value_format = parameters.get("value_format")
+ value = value_format.format(value) # type: ignore
+ return key, operator, value
+
+ return key, "=", value
+
  def select(self, order_by: str = "", limit: int = 0, **kwargs) -> sqlite3.Cursor:
  if kwargs.values():
  column_list = []
  values = []
  for key, value in kwargs.items():
- if value is None:
- where = f"{key} IS NULL"
- else:
- values.append(value)
- where = f"{key} = ?"
+ key, operator, value = self.get_special_key_values(key, value)
+ where = f"{key} {operator} ?"
+ values.append(value)
+
  column_list.append(where)
 
  columns = " AND ".join(column_list)

setup.py

@@ -1,15 +1,15 @@
-# Third Party
+# First Party
 from setuptools import setup
 
 setup(
  name="minislite",
- version="0.91",
+ version="0.99",
  packages=["minislite"],
  url="https://github.com/ahmetkotan/minislite",
  license="",
  author="ahmetkotan",
  author_email="[email protected]",
- description="Mini SQLite ORM",
+ description="Mini and Secure SQLite ORM",
  long_description=open("README.md").read(),
  long_description_content_type="text/markdown",
  classifiers=[

tests/test_objects.py

@@ -2,11 +2,10 @@
 import pytest
 from tests.test_models import MiniModel, ExtendedModel
 from minislite.minislite import MiniSLiteDb
-from minislite.exceptions import AreYouSure, AlreadyExistsError, RecordNotFoundError
+from minislite.exceptions import AreYouSureError, AlreadyExistsError, RecordNotFoundError
 
 
 def test_objects_dunder_get(extended_model: ExtendedModel):
- assert extended_model.objects.table_name == "extended_model"
  assert extended_model.objects.klass == extended_model
  assert extended_model.objects.unique_together == ["name", "last_name"]
  assert extended_model.objects.manager
@@ -135,7 +134,7 @@ def test_objects_delete(extended_model: ExtendedModel, database: MiniSLiteDb):
  extended_model.objects.create(name="extended", last_name="model2", age=2)
  extended_model.objects.create(name="extended", last_name="model3", age=3)
 
- with pytest.raises(AreYouSure):
+ with pytest.raises(AreYouSureError):
  extended_model.objects.delete()
 
  extended_model.objects.delete(object_id=extended1.id)

tests/test_tables.py

@@ -6,28 +6,42 @@
 from minislite.models import MiniSLiteModel
 from minislite.tables import TableManager
 from minislite.minislite import MiniSLiteDb
-from minislite.exceptions import DatabaseNotFound, RecordNotFoundError
+from minislite.exceptions import DatabaseNotFoundError, RecordNotFoundError, WhereOperatorError
 
 
 def test_tables_initialize(extended_model: MiniSLiteModel):
  os.environ.pop("MINISLITE_DB_PATH", None)
 
- with pytest.raises(DatabaseNotFound):
- TableManager(table_name="x")
+ with pytest.raises(DatabaseNotFoundError):
+ TableManager(model=extended_model)
 
  database = MiniSLiteDb("test.db")
  database.add_model(extended_model)
 
- TableManager(extended_model.get_table_name())
+ TableManager(extended_model)
  assert database.connection
  assert database.cursor
 
 
+def test_tables_get_special_key_values(extended_model: MiniSLiteModel, database: MiniSLiteDb):
+ database.add_model(extended_model)
+
+ table = TableManager(extended_model)
+ key, operator, value = table.get_special_key_values("name__contains", "extended")
+
+ assert key == "name"
+ assert operator == "LIKE"
+ assert value == "%extended%"
+
+ with pytest.raises(WhereOperatorError):
+ table.get_special_key_values("name__gt", "extended")
+
+
 def test_tables_select(extended_model: MiniSLiteModel, database: MiniSLiteDb):
  database.add_model(extended_model)
  database.clean_tables()
 
- table = TableManager(extended_model.get_table_name())
+ table = TableManager(extended_model)
 
  extended = extended_model.objects.create(name="extended", last_name="model1", age=None)
  extended_model.objects.create(name="extended", last_name="model2", age=2)
@@ -50,11 +64,29 @@ def test_tables_select(extended_model: MiniSLiteModel, database: MiniSLiteDb):
  assert dict(last_extended).get("last_name") == "model3"
 
 
+def test_tables_select_with_operators(extended_model: MiniSLiteModel, database: MiniSLiteDb):
+ database.add_model(extended_model)
+ database.clean_tables()
+
+ extended_model.objects.create(name="extended", last_name="operator1", age=10)
+ extended_model.objects.create(name="extended", last_name="operator2", age=20)
+ extended_model.objects.create(name="extended", last_name="operator3", age=30)
+ extended_model.objects.create(name="extended2", last_name="notoperator4", age=None)
+
+ table = TableManager(extended_model)
+ assert len(table.select(last_name__contains="operator").fetchall()) == 4
+ assert len(table.select(last_name__startswith="operator").fetchall()) == 3
+ assert len(table.select(last_name__endswith="4").fetchall()) == 1
+
+ assert len(table.select(age__gt=20).fetchall()) == 1
+ assert len(table.select(age__gte=20).fetchall()) == 2
+
+
 def test_tables_insert(extended_model: MiniSLiteModel, database: MiniSLiteDb):
  database.add_model(extended_model)
  database.clean_tables()
 
- table = TableManager(extended_model.get_table_name())
+ table = TableManager(extended_model)
  table.insert(name="extended", last_name="model1", age=1)
 
  assert len(extended_model.objects.all()) == 1
@@ -64,7 +96,7 @@ def test_tables_update(extended_model: MiniSLiteModel, database: MiniSLiteDb):
  database.add_model(extended_model)
  database.clean_tables()
 
- table = TableManager(extended_model.get_table_name())
+ table = TableManager(extended_model)
  assert table.update() is None
 
  extended = extended_model.objects.create(name="extended", last_name="model2", age=2)
@@ -84,7 +116,7 @@ def test_tables_delete(extended_model: MiniSLiteModel, database: MiniSLiteDb):
  database.add_model(extended_model)
  database.clean_tables()
 
- table = TableManager(extended_model.get_table_name())
+ table = TableManager(extended_model)
 
  extended = extended_model.objects.create(name="extended", last_name="model2", age=2)
  extended_model.objects.create(name="extended", last_name="model3", age=3)