From bb02b110c2df65da46b34313181f2470a0ab8b33 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Ege=20Emir=20=C3=96zkan?= Date: Mon, 3 Aug 2020 03:11:22 +0300 Subject: [PATCH] Changed name sqlify to datalite --- setup.py | 0 src/datalite.py | 27 +++++++++++++++++++++------ 2 files changed, 21 insertions(+), 6 deletions(-) create mode 100644 setup.py diff --git a/setup.py b/setup.py new file mode 100644 index 0000000..e69de29 diff --git a/src/datalite.py b/src/datalite.py index cd02461..0f2692e 100644 --- a/src/datalite.py +++ b/src/datalite.py @@ -1,10 +1,11 @@ -from os.path import exists -from pathlib import Path import sqlite3 as sql from dataclasses import Field, asdict, dataclass from typing import List, Dict, Optional, Callable, Any + + + def _convert_type(type_: Optional[type], type_overload: Dict[Optional[type], str]) -> str: """ Given a Python type, return the str name of its @@ -12,6 +13,8 @@ def _convert_type(type_: Optional[type], type_overload: Dict[Optional[type], str :param type_: A Python type, or None. :param type_overload: A type table to overload the custom type table. :return: The str name of the sql type. + >>> _convert_type(int) + "INTEGER" """ try: return type_overload[type_] @@ -77,7 +80,6 @@ def _create_entry(self) -> None: Given an object, create the entry for the object. As a side-effect, this will set the object_id attribute of the object to the unique id of the entry. - :param cur: Cursor of the database. :param self: Instance of the object. :return: None. """ @@ -93,8 +95,20 @@ def _create_entry(self) -> None: con.commit() -def sqlify(db_path: str, type_overload: Optional[Dict[Optional[type], str]] = None, - *args, **kwargs) -> Callable: +def _remove_entry(self) -> None: + """ + Remove the object's record in the underlying database. + :param self: self instance. + :return: None. + """ + with sql.connect(getattr(self, "db_path")) as con: + cur: sql.Cursor = con.cursor() + cur.execute(f"DELETE FROM {self.__class__.__name__.lower()} WHERE obj_id = {self.obj_id}") + con.commit() + + +def datalite(db_path: str, type_overload: Optional[Dict[Optional[type], str]] = None, + *args, **kwargs) -> Callable: def decorator(dataclass_: type, *args_i, **kwargs_i): type_table: Dict[Optional[type], str] = {None: "NULL", int: "INTEGER", float: "REAL", str: "TEXT", bytes: "BLOB"} @@ -109,6 +123,7 @@ def sqlify(db_path: str, type_overload: Optional[Dict[Optional[type], str]] = No return decorator + def is_fetchable(class_: type, obj_id: int) -> bool: """ Check if a record is fetchable given its obj_id and @@ -196,4 +211,4 @@ def fetch_all(class_: type) -> tuple: obj = class_(**kwargs) setattr(obj, "obj_id", obj_id) objects.append(obj) - return tuple(objects) \ No newline at end of file + return tuple(objects)