fiableDB/fiable_db.py

205 lines
5.9 KiB
Python
Raw Normal View History

2022-11-21 16:37:43 +01:00
from typing import Dict, Tuple, Union, Sequence, TypedDict
2022-11-30 11:05:31 +01:00
from functools import reduce
2022-11-28 16:10:49 +01:00
import json
from os import path
2022-11-21 16:37:43 +01:00
# Variables
2022-11-09 12:26:06 +01:00
FILE = "fiabledb.json"
2022-11-28 16:10:49 +01:00
database = []
2022-11-21 16:37:43 +01:00
# Type aliases
2022-11-28 16:10:49 +01:00
2022-11-21 16:37:43 +01:00
class TypeData(TypedDict):
id: int
rev: int
data: dict
Type_Data_List = Tuple[TypeData]
Type_Add_Data = Union[Dict, Sequence[Dict]]
2022-11-28 16:50:21 +01:00
Type_Add_Return = Union[Tuple[int, int, Dict], Tuple[Tuple[int, int, Dict]], None]
2022-12-05 15:15:45 +01:00
Type_Update_Return = Union[Tuple[Tuple[int, int, Dict]], None]
Type_Delete_Return = Union[Tuple[Tuple[int, int, Dict]], None]
2022-11-21 16:37:43 +01:00
Type_Find_One_Return = TypeData
Type_Find_All_Return = Tuple[Type_Find_One_Return]
2022-11-09 16:59:42 +01:00
2022-11-28 16:10:49 +01:00
# Functions
2022-11-28 16:50:21 +01:00
2022-11-28 16:10:49 +01:00
def get_next_id(table: str = "default") -> int:
2022-11-30 11:05:31 +01:00
"""Get the next id for the table"""
2022-11-28 16:10:49 +01:00
global database
2022-11-30 11:05:31 +01:00
# Get the last id for the table
def get_id(current_id, row: TypeData) -> int:
if current_id == None and table == row["table"]:
return row["id"]
else:
return current_id
2022-11-30 12:57:16 +01:00
2022-11-30 11:05:31 +01:00
last_id = reduce(get_id, database[::-1], None)
# Return the next id, or 1 if there is no last id
return last_id + 1 if last_id else 1
2022-11-28 16:10:49 +01:00
2022-11-21 16:37:43 +01:00
def start(file_name: str = "") -> str:
2022-11-10 18:54:16 +01:00
"""Start the database
Args:
file (str, optional): The file to use. Defaults to FILE.
Returns:
str: The file used
"""
2022-11-28 16:50:21 +01:00
global FILE
2022-11-21 16:37:43 +01:00
global database
2022-11-28 16:50:21 +01:00
my_file_name = file_name if file_name != "" else FILE
2022-11-21 16:37:43 +01:00
if path.exists(my_file_name):
# Load the database
load(my_file_name)
else:
# Create the database
2022-11-28 16:50:21 +01:00
database = []
save(my_file_name)
2022-11-21 16:37:43 +01:00
return my_file_name
2022-11-10 18:54:16 +01:00
2022-11-28 16:50:21 +01:00
def save(file_name: str = "") -> bool:
2022-11-10 19:00:57 +01:00
"""Save the database
Args:
2022-11-21 16:37:43 +01:00
file_name (str, optional): The file to save to. Defaults to "".
2022-11-10 19:00:57 +01:00
data (list[str, list[int, int, dict]], optional): The data to save. Defaults to {}.
Returns:
bool: True if the data was saved, False otherwise
"""
2022-11-28 16:50:21 +01:00
global FILE
2022-11-21 16:37:43 +01:00
global database
2022-11-28 16:50:21 +01:00
my_file_name = file_name if file_name != "" else FILE
2022-11-21 16:37:43 +01:00
with open(my_file_name, "w") as f:
2022-11-28 16:50:21 +01:00
json.dump(database, f)
2022-11-21 16:37:43 +01:00
return True
2022-11-10 19:00:57 +01:00
2022-11-21 16:37:43 +01:00
def load(file_name: Union[str, None] = None) -> bool:
2022-11-10 19:00:57 +01:00
"""Load the database
Args:
2022-11-21 16:37:43 +01:00
file_name (str, optional): The file to load from. Defaults to "".
2022-11-10 19:00:57 +01:00
Returns:
2022-11-21 16:37:43 +01:00
Bool - The data loaded
2022-11-10 19:00:57 +01:00
"""
2022-11-21 16:37:43 +01:00
global database
my_file_name = file_name if file_name else FILE
is_exists = path.exists(my_file_name)
if is_exists:
with open(my_file_name, "r") as f:
text = f.read()
if text != "":
database = json.loads(text)
else:
database = []
else:
raise FileNotFoundError("File not found")
return is_exists
def get_database() -> Type_Data_List:
"""Get the data
Returns:
list[dict[int, int, dict]]: The data
"""
global database
return database
2022-11-10 19:00:57 +01:00
2022-11-28 16:10:49 +01:00
def add(new_data: Type_Add_Data, table: str = "default") -> Type_Add_Return:
2022-11-10 18:54:16 +01:00
"""Add data to the database
Args:
new_data (dict|list): The data to add
table (str, optional): The table to add to. Defaults to "".
Returns:
dict[int, int, dict]|list[dict[int, int, dict]]: The data added
"""
2022-11-28 16:10:49 +01:00
global database
2022-11-10 18:54:16 +01:00
if isinstance(new_data, dict):
2022-11-30 09:52:21 +01:00
new_row = {"id": get_next_id(table), "rev": 1, "table": table, "data": new_data}
2022-11-28 16:10:49 +01:00
database.append(new_row)
return new_row
2022-11-10 18:54:16 +01:00
elif isinstance(new_data, list):
2022-11-28 16:10:49 +01:00
for row in new_data:
2022-11-30 09:52:21 +01:00
new_row = {"id": get_next_id(table), "rev": 1, "table": table, "data": row}
2022-11-28 16:10:49 +01:00
database.append(new_row)
2022-11-10 18:54:16 +01:00
else:
raise TypeError("new_data must be a dict or list")
def update(
2022-12-05 14:52:09 +01:00
id: int, new_data: dict, table: str = "default", force: bool = False
2022-11-21 16:37:43 +01:00
) -> Type_Update_Return:
2022-11-10 18:54:16 +01:00
"""Update data in the database
Args:
2022-11-13 16:04:24 +01:00
id (int): The id of the data to update.
new_data (dict): The data to update
2022-11-10 18:54:16 +01:00
table (str, optional): The table to update. Defaults to "".
force (bool, optional): Force the update. Defaults to False.
Returns:
2022-12-05 14:52:09 +01:00
dict[int, int, dict] or None: The data updated
2022-11-10 18:54:16 +01:00
"""
2022-12-05 14:52:09 +01:00
global database
# Get the key to update
def get_key(
current_key: Union[Tuple[int, int, Dict], None], key_and_row: TypeData
) -> Union[int, None]:
if (
current_key == None
and key_and_row["id"] == id
and key_and_row["table"] == table
):
return key_and_row[0]
return None
key = reduce(get_key, list(enumerate(database))[::-1], None)
if key:
if force:
database[key]["data"] = new_data
else:
database[key]["data"].update(new_data)
database[key]["rev"] += 1
return None
2022-11-10 18:54:16 +01:00
2022-11-21 16:37:43 +01:00
def delete(id: int, data: dict, table: str = "") -> Type_Delete_Return:
2022-11-10 18:54:16 +01:00
"""Delete data from the database
Args:
id (int): The id of the data to delete
data (dict): Filter the data to delete
table (str, optional): The table to delete from. Defaults to "".
Returns:
dict[int, int, dict]: The data deleted
"""
2022-11-13 15:13:43 +01:00
print("Function not implemented yet")
2022-11-10 18:54:16 +01:00
def find_one(
id: int = 0, data: dict = {}, table: str = "", rev: int = 0
2022-11-21 16:37:43 +01:00
) -> Type_Find_One_Return:
2022-11-10 18:54:16 +01:00
"""Find one data in the database
Args:
id (int, optional): The id of the data to find. Defaults to 0.
data (dict, optional): Filter the data to find. Defaults to {}.
table (str, optional): The table to find in. Defaults to "".
rev (int, optional): The revision of the data to find. Defaults to 0.
Returns:
dict[int, int, dict]: The data found
"""
2022-11-13 15:13:43 +01:00
print("Function not implemented yet")
2022-11-10 18:54:16 +01:00
2022-11-21 16:37:43 +01:00
def find_all(data: dict = {}, table: str = "") -> Type_Find_All_Return:
2022-11-10 18:54:16 +01:00
"""Find all data in the database
Args:
data (dict, optional): Filter the data to find. Defaults to {}.
table (str, optional): The table to find in. Defaults to "".
Returns:
list[dict[int, int, dict]]: The data found
"""
2022-11-13 15:13:43 +01:00
print("Function not implemented yet")