Verified Commit 19446bf0 authored by Dmitry Volodin's avatar Dmitry Volodin
Browse files

datastream: administrativedomain datastream

parent c2ca0868
......@@ -8,10 +8,12 @@
# NOC modules
from noc.sa.models.managedobject import ManagedObject
from noc.sa.models.administrativedomain import AdministrativeDomain
from noc.core.datastream.change import bulk_datastream_changes
def fix():
with bulk_datastream_changes():
for mo in ManagedObject.objects.all():
mo.save() # Force datastream rebuild
for model in [AdministrativeDomain, ManagedObject]:
with bulk_datastream_changes():
for obj in model.objects.all():
obj.save() # Force datastream rebuild
......@@ -2,7 +2,7 @@
# ----------------------------------------------------------------------
# AdministrativeDomain
# ----------------------------------------------------------------------
# Copyright (C) 2007-2017 The NOC Project
# Copyright (C) 2007-2018 The NOC Project
# See LICENSE for details
# ----------------------------------------------------------------------
......@@ -19,12 +19,14 @@ from noc.main.models.remotesystem import RemoteSystem
from noc.core.model.fields import TagsField, DocumentReferenceField
from noc.core.model.decorator import on_delete_check
from noc.core.bi.decorator import bi_sync
from noc.core.datastream.decorator import datastream
id_lock = Lock()
@bi_sync
@datastream
@on_delete_check(check=[
("cm.ObjectNotify", "administrative_domain"),
# ("fm.EscalationItem", "administrative_domain"),
......@@ -91,6 +93,9 @@ class AdministrativeDomain(models.Model):
else:
return None
def iter_changed_datastream(self):
yield "administrativedomain", self.id
@cachetools.cachedmethod(operator.attrgetter("_path_cache"), lock=lambda _: id_lock)
def get_path(self):
"""
......
# -*- coding: utf-8 -*-
# ----------------------------------------------------------------------
# administrativedomain datastream
# ----------------------------------------------------------------------
# Copyright (C) 2007-2018 The NOC Project
# See LICENSE for details
# ----------------------------------------------------------------------
# NOC modules
from noc.core.datastream.base import DataStream
from noc.sa.models.administrativedomain import AdministrativeDomain
def qs(s):
if not s:
return ""
return s.encode("utf-8")
class AdmDomainDataStream(DataStream):
name = "administrativedomain"
clean_id = DataStream.clean_id_int
@classmethod
def get_object(cls, id):
ad = AdministrativeDomain.objects.filter(id=id)[:1]
if not ad:
raise KeyError()
ad = ad[0]
r = {
"id": str(ad.id),
"name": qs(ad.name),
}
if ad.parent:
r["parent"] = str(ad.parent.id)
if ad.tags:
r["tags"] = [qs(t) for t in ad.tags]
cls._apply_remote_system(ad, r)
return r
@staticmethod
def _apply_remote_system(mo, r):
if mo.remote_system:
r["remote_system"] = {
"id": str(mo.remote_system.id),
"name": qs(mo.remote_system.name)
}
r["remote_id"] = mo.remote_id
......@@ -266,7 +266,7 @@ def test_datastream_clean_id_int():
DS.clean_id("z")
@pytest.fixture(params=["managedobject"])
@pytest.fixture(params=["managedobject", "administrativedomain"])
def datastream_name(request):
return request.param
......
Supports Markdown
0% or .
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment