Commit be7f4145 authored by Lluis Gifre Renom's avatar Lluis Gifre Renom
Browse files

Intermediate Backup: 

- Extended database API with novel ORM
- Carrying out tests over ORM and InMemoryDB
- Foreign Keys need more testing
parent 9cf957d0

report_coverage_common.sh

0 → 100755
+3 −0
Original line number Diff line number Diff line 
#!/bin/bash



./report_coverage_all.sh | grep --color -E -i "^common/.*$|$"

report_coverage_common_orm.sh

0 → 100755
+3 −0
Original line number Diff line number Diff line 
#!/bin/bash



./report_coverage_all.sh | grep -v -E "^(cent|comp|cont|devi|moni|serv|test)" | grep --color -E -i "^common/orm/.*$|$"

src/common/orm/_Database.py

+7 −9
Original line number Diff line number Diff line 
import logging

from typing import List, Set

from .backend._Backend import _Backend

from .fields.PrimaryKeyField import PrimaryKeyField

from .model.Model import Model

from .Exceptions import MutexException
@@ -26,17 +25,16 @@ class _Database(Model): 
    @property

    def backend_key(self) -> str: return ''



    def __enter__(self) -> '_Database':

        self._acquired, self._owner_key = self._backend.lock()

        if not self._acquired: raise MutexException('Unable to acquire database lock')

        return self

    #def __enter__(self) -> '_Database':

    #    self._acquired, self._owner_key = self._backend.lock()

    #    if not self._acquired: raise MutexException('Unable to acquire database lock')

    #    return self



    def __exit__(self, exc_type, exc_val, exc_tb) -> None:

        self._backend.unlock(self._owner_key)

    #def __exit__(self, exc_type, exc_val, exc_tb) -> None:

    #    self._backend.unlock(self._owner_key)



    def clear_all(self, keep_keys : Set[str] = set()) -> None:

        keys = self._backend.keys()

        for key in keys:

        for key in self._backend.keys():

            if key in keep_keys: continue

            self._backend.delete(key)

src/common/orm/backend/Tools.py

0 → 100644
+5 −0
Original line number Diff line number Diff line 
from typing import List, Union



def key_to_str(key : Union[str, List[str]]) -> str:

    if isinstance(key, str): return key

    return '/'.join(map(str, key))

src/common/orm/backend/_Backend.py

+14 −14
Original line number Diff line number Diff line 
from typing import Dict, List, Set, Tuple

from typing import Dict, List, Optional, Tuple



class _Backend:

    def __init__(self, **settings) -> None:

        raise NotImplementedError()



    def lock(self) -> Tuple[bool, str]:

    def lock(self, keys : List[List[str]], owner_key : Optional[str] = None) -> Tuple[bool, str]:

        raise NotImplementedError()



    def unlock(self, owner_key : str) -> bool:

    def unlock(self, keys : List[List[str]], owner_key : str) -> bool:

        raise NotImplementedError()



    def keys(self) -> list:

        raise NotImplementedError()



    def exists(self, key_name : str) -> bool:

    def exists(self, key : List[str]) -> bool:

        raise NotImplementedError()



    def delete(self, key_name : str) -> bool:

    def delete(self, key : List[str]) -> bool:

        raise NotImplementedError()



    def dict_get(self, key_name : str, fields : List[str] = []) -> Dict[str, str]:

    def dict_get(self, key : List[str], fields : List[str] = []) -> Dict[str, str]:

        raise NotImplementedError()



    def dict_update(self, key_name : str, update_fields : Dict[str,str] = {}, remove_fields : Set[str] = set()) -> None:

    def dict_update(self, key : List[str], fields : Dict[str,str] = {}) -> None:

        raise NotImplementedError()



    def dict_delete(self, key_name : str, fields : List[str] = []) -> None:

    def dict_delete(self, key : List[str], fields : List[str] = []) -> None:

        raise NotImplementedError()



    def list_get_all(self, key_name : str) -> List[str]:

    def list_get_all(self, key : List[str]) -> List[str]:

        raise NotImplementedError()



    def list_push_last(self, key_name : str, item : str) -> None:

    def list_push_last(self, key : List[str], item : str) -> None:

        raise NotImplementedError()



    def list_remove_first_occurrence(self, key_name : str, item: str) -> None:

    def list_remove_first_occurrence(self, key : List[str], item: str) -> None:

        raise NotImplementedError()



    def set_add(self, key_name : str, item : str) -> None:

    def set_add(self, key : List[str], item : str) -> None:

        raise NotImplementedError()



    def set_has(self, key_name : str, item : str) -> bool:

    def set_has(self, key : List[str], item : str) -> bool:

        raise NotImplementedError()



    def set_remove(self, key_name : str, item : str) -> None:

    def set_remove(self, key : List[str], item : str) -> None:

        raise NotImplementedError()



    def dump(self) -> List[str]: