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

Context component: 

- relocated database methods
- corrected models to use single-column primary key
- corrected test cases
parent d649fe78

src/context/service/ContextServiceServicerImpl.py

+26 −29
Original line number Diff line number Diff line
@@ -35,14 +35,11 @@ from common.proto.context_policy_pb2_grpc import ContextPolicyServiceServicer 
from common.rpc_method_wrapper.Decorator import create_metrics, safe_and_metered_rpc_method

#from common.rpc_method_wrapper.ServiceExceptions import (

#    InvalidArgumentException, NotFoundException, OperationFailedException)

from .database.methods.Context import (

    context_delete, context_get, context_list_ids, context_list_objs, context_set)

from .database.methods.Device import (

    device_delete, device_get, device_list_ids, device_list_objs, device_set)

#from .database.methods.Link import link_delete, link_get, link_list_ids, link_list_objs, link_set

#from .database.methods.Service import service_delete, service_get, service_list_ids, service_list_objs, service_set

from .database.methods.Topology import (

    topology_delete, topology_get, topology_list_ids, topology_list_objs, topology_set)

from .database.Context import context_delete, context_get, context_list_ids, context_list_objs, context_set

from .database.Device import device_delete, device_get, device_list_ids, device_list_objs, device_set

from .database.Link import link_delete, link_get, link_list_ids, link_list_objs, link_set

#from .database.Service import service_delete, service_get, service_list_ids, service_list_objs, service_set

from .database.Topology import topology_delete, topology_get, topology_list_ids, topology_list_objs, topology_set

#from common.tools.grpc.Tools import grpc_message_to_json, grpc_message_to_json_string

#from context.service.Database import Database

#from context.service.database.ConfigModel import (
@@ -200,31 +197,31 @@ class ContextServiceServicerImpl(ContextServiceServicer, ContextPolicyServiceSer 


    # ----- Link -------------------------------------------------------------------------------------------------------



#    @safe_and_metered_rpc_method(METRICS, LOGGER)

#    def ListLinkIds(self, request : Empty, context : grpc.ServicerContext) -> LinkIdList:

#        return link_list_ids(self.db_engine)

    @safe_and_metered_rpc_method(METRICS, LOGGER)

    def ListLinkIds(self, request : Empty, context : grpc.ServicerContext) -> LinkIdList:

        return link_list_ids(self.db_engine)



#    @safe_and_metered_rpc_method(METRICS, LOGGER)

#    def ListLinks(self, request : Empty, context : grpc.ServicerContext) -> LinkList:

#        return link_list_objs(self.db_engine)

    @safe_and_metered_rpc_method(METRICS, LOGGER)

    def ListLinks(self, request : Empty, context : grpc.ServicerContext) -> LinkList:

        return link_list_objs(self.db_engine)



#    @safe_and_metered_rpc_method(METRICS, LOGGER)

#    def GetLink(self, request : LinkId, context : grpc.ServicerContext) -> Link:

#        return link_get(self.db_engine, request)

    @safe_and_metered_rpc_method(METRICS, LOGGER)

    def GetLink(self, request : LinkId, context : grpc.ServicerContext) -> Link:

        return link_get(self.db_engine, request)



#    @safe_and_metered_rpc_method(METRICS, LOGGER)

#    def SetLink(self, request : Link, context : grpc.ServicerContext) -> LinkId:

#        link_id,updated = link_set(self.db_engine, request)

#        #event_type = EventTypeEnum.EVENTTYPE_UPDATE if updated else EventTypeEnum.EVENTTYPE_CREATE

#        #notify_event(self.messagebroker, TOPIC_LINK, event_type, {'link_id': link_id})

#        return link_id

    @safe_and_metered_rpc_method(METRICS, LOGGER)

    def SetLink(self, request : Link, context : grpc.ServicerContext) -> LinkId:

        link_id,updated = link_set(self.db_engine, request)

        #event_type = EventTypeEnum.EVENTTYPE_UPDATE if updated else EventTypeEnum.EVENTTYPE_CREATE

        #notify_event(self.messagebroker, TOPIC_LINK, event_type, {'link_id': link_id})

        return link_id



#    @safe_and_metered_rpc_method(METRICS, LOGGER)

#    def RemoveLink(self, request : LinkId, context : grpc.ServicerContext) -> Empty:

#        deleted = link_delete(self.db_engine, request)

#        #if deleted:

#        #    notify_event(self.messagebroker, TOPIC_LINK, event_type, {'link_id': dict_link_id})

#        return Empty()

    @safe_and_metered_rpc_method(METRICS, LOGGER)

    def RemoveLink(self, request : LinkId, context : grpc.ServicerContext) -> Empty:

        deleted = link_delete(self.db_engine, request)

        #if deleted:

        #    notify_event(self.messagebroker, TOPIC_LINK, EventTypeEnum.EVENTTYPE_REMOVE, {'link_id': request})

        return Empty()



    @safe_and_metered_rpc_method(METRICS, LOGGER)

    def GetLinkEvents(self, request : Empty, context : grpc.ServicerContext) -> Iterator[LinkEvent]:

src/context/service/database/methods/Context.py→src/context/service/database/Context.py

+1 −1
Original line number Diff line number Diff line
@@ -21,7 +21,7 @@ from typing import Dict, List, Optional, Tuple 
from common.proto.context_pb2 import Context, ContextId, ContextIdList, ContextList

from common.rpc_method_wrapper.ServiceExceptions import NotFoundException

from common.tools.object_factory.Context import json_context_id

from context.service.database.models.ContextModel import ContextModel

from .models.ContextModel import ContextModel

from .uuids.Context import context_get_uuid



LOGGER = logging.getLogger(__name__)

src/context/service/database/methods/Device.py→src/context/service/database/Device.py

+29 −24
Original line number Diff line number Diff line
@@ -21,15 +21,16 @@ from typing import Dict, List, Optional, Set, Tuple 
from common.proto.context_pb2 import Device, DeviceId, DeviceIdList, DeviceList

from common.rpc_method_wrapper.ServiceExceptions import InvalidArgumentException, NotFoundException

from common.tools.object_factory.Device import json_device_id

#from common.tools.grpc.Tools import grpc_message_to_json_string

#from context.service.database.models.ConfigRuleModel import ConfigRuleKindEnum, ConfigRuleModel

from context.service.database.models.DeviceModel import DeviceModel

from context.service.database.models.EndPointModel import EndPointModel

from context.service.database.models.RelationModels import TopologyDeviceModel

#from context.service.database.models.enums.ConfigAction import grpc_to_enum__config_action

from context.service.database.models.enums.DeviceDriver import grpc_to_enum__device_driver

from context.service.database.models.enums.DeviceOperationalStatus import grpc_to_enum__device_operational_status

from context.service.database.models.enums.KpiSampleType import grpc_to_enum__kpi_sample_type

from common.tools.grpc.Tools import grpc_message_to_json_string

from .models.ConfigRuleModel import ConfigRuleKindEnum, ConfigRuleModel

from .models.DeviceModel import DeviceModel

from .models.EndPointModel import EndPointModel

from .models.RelationModels import TopologyDeviceModel

from .models.enums.ConfigAction import grpc_to_enum__config_action

from .models.enums.DeviceDriver import grpc_to_enum__device_driver

from .models.enums.DeviceOperationalStatus import grpc_to_enum__device_operational_status

from .models.enums.KpiSampleType import grpc_to_enum__kpi_sample_type

from .uuids._Builder import get_uuid_random

from .uuids.Device import device_get_uuid

from .uuids.EndPoint import endpoint_get_uuid
@@ -64,7 +65,7 @@ def device_get(db_engine : Engine, request : DeviceId) -> Device: 
def device_set(db_engine : Engine, request : Device) -> bool:

    raw_device_uuid = request.device_id.device_uuid.uuid

    raw_device_name = request.name

    device_name = request.device_id.device_uuid.uuid if len(raw_device_name) == 0 else raw_device_name

    device_name = raw_device_uuid if len(raw_device_name) == 0 else raw_device_name

    device_uuid = device_get_uuid(request.device_id, device_name=device_name, allow_random=True)



    device_type = request.device_type
@@ -83,9 +84,11 @@ def device_set(db_engine : Engine, request : Device) -> bool: 
                ['should be == request.device_id.device_uuid.uuid({:s})'.format(raw_device_uuid)]

            )



        raw_endpoint_uuid = endpoint.endpoint_id.endpoint_uuid.uuid

        raw_endpoint_name = endpoint.name

        endpoint_topology_uuid, endpoint_device_uuid, endpoint_uuid = endpoint_get_uuid(

            endpoint.endpoint_id, endpoint_name=raw_endpoint_name, allow_random=True)

        endpoint_name = raw_endpoint_uuid if len(raw_endpoint_name) == 0 else raw_endpoint_name



        kpi_sample_types = [grpc_to_enum__kpi_sample_type(kst) for kst in endpoint.kpi_sample_types]
@@ -93,7 +96,7 @@ def device_set(db_engine : Engine, request : Device) -> bool: 
            'endpoint_uuid'   : endpoint_uuid,

            'device_uuid'     : endpoint_device_uuid,

            'topology_uuid'   : endpoint_topology_uuid,

            'name'            : raw_endpoint_name,

            'name'            : endpoint_name,

            'endpoint_type'   : endpoint.endpoint_type,

            'kpi_sample_types': kpi_sample_types,

        })
@@ -101,20 +104,22 @@ def device_set(db_engine : Engine, request : Device) -> bool: 
        if endpoint_topology_uuid not in topology_uuids:

            related_topologies.append({

                'topology_uuid': endpoint_topology_uuid,

                'device_uuid'  : endpoint_device_uuid,

                'device_uuid'  : device_uuid,

            })

            topology_uuids.add(endpoint_topology_uuid)



    #config_rules : List[Dict] = list()

    #for position,config_rule in enumerate(request.device_config.config_rules):

    #    str_kind = config_rule.WhichOneof('config_rule')

    #    config_rules.append({

    #        'device_uuid': device_uuid,

    #        'kind'       : ConfigRuleKindEnum._member_map_.get(str_kind.upper()), # pylint: disable=no-member

    #        'action'     : grpc_to_enum__config_action(config_rule.action),

    #        'position'   : position,

    #        'data'       : grpc_message_to_json_string(getattr(config_rule, str_kind, {})),

    #    })

    config_rules : List[Dict] = list()

    for position,config_rule in enumerate(request.device_config.config_rules):

        configrule_uuid = get_uuid_random()

        str_kind = config_rule.WhichOneof('config_rule')

        config_rules.append({

            'configrule_uuid': configrule_uuid,

            'device_uuid'    : device_uuid,

            'position'       : position,

            'kind'           : ConfigRuleKindEnum._member_map_.get(str_kind.upper()), # pylint: disable=no-member

            'action'         : grpc_to_enum__config_action(config_rule.action),

            'data'           : grpc_message_to_json_string(getattr(config_rule, str_kind, {})),

        })



    device_data = [{

        'device_uuid'              : device_uuid,
@@ -152,8 +157,8 @@ def device_set(db_engine : Engine, request : Device) -> bool: 
            index_elements=[TopologyDeviceModel.topology_uuid, TopologyDeviceModel.device_uuid]

        ))



        #session.execute(delete(ConfigRuleModel).where(ConfigRuleModel.device_uuid == device_uuid))

        #session.execute(insert(ConfigRuleModel).values(config_rules))

        session.execute(delete(ConfigRuleModel).where(ConfigRuleModel.device_uuid == device_uuid))

        session.execute(insert(ConfigRuleModel).values(config_rules))



    run_transaction(sessionmaker(bind=db_engine), callback)

    updated = False # TODO: improve and check if created/updated

src/context/service/database/methods/Link.py→src/context/service/database/Link.py

+40 −43
Original line number Diff line number Diff line
@@ -12,7 +12,6 @@ 
# See the License for the specific language governing permissions and

# limitations under the License.



import time

from sqlalchemy.dialects.postgresql import insert

from sqlalchemy.engine import Engine

from sqlalchemy.orm import Session, sessionmaker
@@ -20,8 +19,11 @@ from sqlalchemy_cockroachdb import run_transaction 
from typing import Dict, List, Optional, Set, Tuple

from common.proto.context_pb2 import Link, LinkId, LinkIdList, LinkList

from common.rpc_method_wrapper.ServiceExceptions import NotFoundException

from context.service.database.models.LinkModel import LinkModel

from context.service.database.models.RelationModels import LinkEndPointModel, TopologyLinkModel

from common.tools.object_factory.Link import json_link_id

from .models.LinkModel import LinkModel

from .models.RelationModels import LinkEndPointModel, TopologyLinkModel

from .uuids.EndPoint import endpoint_get_uuid

from .uuids.Link import link_get_uuid



def link_list_ids(db_engine : Engine) -> LinkIdList:

    def callback(session : Session) -> List[Dict]:
@@ -38,81 +40,76 @@ def link_list_objs(db_engine : Engine) -> LinkList: 
    return LinkList(links=run_transaction(sessionmaker(bind=db_engine), callback))



def link_get(db_engine : Engine, request : LinkId) -> Link:

    link_uuid = request.link_uuid.uuid

    link_uuid = link_get_uuid(request, allow_random=False)

    def callback(session : Session) -> Optional[Dict]:

        obj : Optional[LinkModel] = session.query(LinkModel)\

            .filter_by(link_uuid=link_uuid).one_or_none()

        return None if obj is None else obj.dump()

    obj = run_transaction(sessionmaker(bind=db_engine), callback)

    if obj is None: raise NotFoundException('Link', link_uuid)

    if obj is None:

        raw_link_uuid = request.link_uuid.uuid

        raise NotFoundException('Link', raw_link_uuid, extra_details=[

            'link_uuid generated was: {:s}'.format(link_uuid)

        ])

    return Link(**obj)



def link_set(db_engine : Engine, request : Link) -> bool:

    link_uuid = request.link_id.link_uuid.uuid

    link_name = request.name

    raw_link_uuid = request.link_id.link_uuid.uuid

    raw_link_name = request.name

    link_name = raw_link_uuid if len(raw_link_name) == 0 else raw_link_name

    link_uuid = link_get_uuid(request.link_id, link_name=link_name, allow_random=True)



    topology_keys : Set[Tuple[str, str]] = set()

    topology_uuids : Set[str] = set()

    related_topologies : List[Dict] = list()

    link_endpoints_data : List[Dict] = list()

    for endpoint_id in request.link_endpoint_ids:

        context_uuid  = endpoint_id.topology_id.context_id.context_uuid.uuid

        topology_uuid = endpoint_id.topology_id.topology_uuid.uuid

        device_uuid   = endpoint_id.device_id.device_uuid.uuid

        endpoint_uuid = endpoint_id.endpoint_uuid.uuid

        endpoint_topology_uuid, _, endpoint_uuid = endpoint_get_uuid(

            endpoint_id, allow_random=False)



        link_endpoints_data.append({

            'link_uuid'    : link_uuid,

            'context_uuid' : context_uuid,

            'topology_uuid': topology_uuid,

            'device_uuid'  : device_uuid,

            'endpoint_uuid': endpoint_uuid,

        })



        if len(context_uuid) > 0 and len(topology_uuid) > 0:

            topology_key = (context_uuid, topology_uuid)

            if topology_key not in topology_keys:

        if endpoint_topology_uuid not in topology_uuids:

            related_topologies.append({

                    'context_uuid': context_uuid,

                    'topology_uuid': topology_uuid,

                'topology_uuid': endpoint_topology_uuid,

                'link_uuid': link_uuid,

            })

                topology_keys.add(topology_key)

            topology_uuids.add(endpoint_topology_uuid)



    link_data = [{

        'link_uuid': link_uuid,

        'link_name': link_name,

    }]



    def callback(session : Session) -> None:

        obj : Optional[LinkModel] = session.query(LinkModel).with_for_update()\

            .filter_by(link_uuid=link_uuid).one_or_none()

        is_update = obj is not None

        if is_update:

            obj.link_name = link_name

            session.merge(obj)

        else:

            session.add(LinkModel(link_uuid=link_uuid, link_name=link_name, created_at=time.time()))

        obj : Optional[LinkModel] = session.query(LinkModel)\

            .filter_by(link_uuid=link_uuid).one_or_none()

        stmt = insert(LinkModel).values(link_data)

        stmt = stmt.on_conflict_do_update(

            index_elements=[LinkModel.link_uuid],

            set_=dict(link_name = stmt.excluded.link_name)

        )

        session.execute(stmt)



        stmt = insert(LinkEndPointModel).values(link_endpoints_data)

        stmt = stmt.on_conflict_do_nothing(

            index_elements=[

                LinkEndPointModel.link_uuid, LinkEndPointModel.context_uuid, LinkEndPointModel.topology_uuid,

                LinkEndPointModel.device_uuid, LinkEndPointModel.endpoint_uuid

            ],

            index_elements=[LinkEndPointModel.link_uuid, LinkEndPointModel.endpoint_uuid]

        )

        session.execute(stmt)



        session.execute(insert(TopologyLinkModel).values(related_topologies).on_conflict_do_nothing(

            index_elements=[

                TopologyLinkModel.context_uuid, TopologyLinkModel.topology_uuid,

                TopologyLinkModel.link_uuid

            ]

            index_elements=[TopologyLinkModel.topology_uuid, TopologyLinkModel.link_uuid]

        ))



    run_transaction(sessionmaker(bind=db_engine), callback)

    return False # TODO: improve and check if created/updated

    updated = False # TODO: improve and check if created/updated

    return LinkId(**json_link_id(link_uuid)),updated



def link_delete(db_engine : Engine, request : LinkId) -> bool:

    link_uuid = request.link_uuid.uuid

    link_uuid = link_get_uuid(request, allow_random=False)

    def callback(session : Session) -> bool:

        session.query(TopologyLinkModel).filter_by(link_uuid=link_uuid).delete()

        session.query(LinkEndPointModel).filter_by(link_uuid=link_uuid).delete()

        #session.query(TopologyLinkModel).filter_by(link_uuid=link_uuid).delete()

        #session.query(LinkEndPointModel).filter_by(link_uuid=link_uuid).delete()

        num_deleted = session.query(LinkModel).filter_by(link_uuid=link_uuid).delete()

        #db_link = session.query(LinkModel).filter_by(link_uuid=link_uuid).one_or_none()

        #session.query(LinkModel).filter_by(link_uuid=link_uuid).delete()

src/context/service/database/methods/Service.py→src/context/service/database/Service.py

+1 −1
Original line number Diff line number Diff line
@@ -20,7 +20,7 @@ from sqlalchemy_cockroachdb import run_transaction 
from typing import Dict, List, Optional

from common.proto.context_pb2 import ContextId, Service, ServiceId, ServiceIdList, ServiceList

from common.rpc_method_wrapper.ServiceExceptions import InvalidArgumentException, NotFoundException

from context.service.database.models.ServiceModel import ServiceModel

from .models.ServiceModel import ServiceModel



def service_list_ids(db_engine : Engine, request : ContextId) -> ServiceIdList:

    context_uuid = request.context_uuid.uuid