Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def _get_items_from_list(items):
new_items = []
for v in items:
value = v
if isinstance(v, dict):
value = _get_items_from_dict(v)
elif isinstance(v, list):
value = _get_items_from_list(v)
elif not isinstance(v, (DBRef, Document)):
value = field.to_python(v)
new_items.append(value)
return new_items
@six.python_2_unicode_compatible
class PrefixListCachePrefix(EmbeddedDocument):
meta = {"strict": False, "auto_create_index": False}
prefix = StringField(required=True)
min = IntField(required=True)
max = IntField(required=True)
def __str__(self):
return self.prefix
@six.python_2_unicode_compatible
class PrefixListCache(Document):
"""
Prepared prefix-list cache. Can hold IPv4/IPv6 prefixes at same time.
Prefixes are stored sorted
"""
meta = {"collection": "noc.prefix_list_cache", "strict": False, "auto_create_index": False}
peering_point = ForeignKeyField(PeeringPoint)
name = StringField()
prefixes = ListField(EmbeddedDocumentField(PrefixListCachePrefix))
changed = DateTimeField()
pushed = DateTimeField()
def __str__(self):
return " %s/%s" % (self.peering_point.hostname, self.name)
# -*- coding: utf-8 -*-
# ----------------------------------------------------------------------
# CHPolicy model
# ----------------------------------------------------------------------
# Copyright (C) 2007-2019 The NOC Project
# See LICENSE for details
# ----------------------------------------------------------------------
# Third-party modules
import six
from mongoengine.document import Document
from mongoengine.fields import StringField, BooleanField, IntField
@six.python_2_unicode_compatible
class CHPolicy(Document):
meta = {"collection": "chpolicies", "strict": False, "auto_create_index": False}
table = StringField(unique=True)
is_active = BooleanField(default=True)
dry_run = BooleanField(default=True)
ttl = IntField(default=0)
def __str__(self):
return self.table
BooleanField,
)
# NOC modules
from .validationpolicy import ValidationPolicy
class ValidationPolicyItem(EmbeddedDocument):
policy = ReferenceField(ValidationPolicy)
is_active = BooleanField(default=True)
def __str__(self):
return self.policy.name
class ValidationPolicySettings(Document):
meta = {
"collection": "noc.validationpolicysettings",
"strict": False,
"auto_create_index": False,
"indexes": [("model_id", "object_id")],
}
model_id = StringField()
object_id = StringField()
policies = ListField(EmbeddedDocumentField(ValidationPolicyItem))
def __str__(self):
return "%s: %s" % (self.model_id, self.object_id)
## See LICENSE for details
##----------------------------------------------------------------------
## Python modules
import threading
import time
import operator
## Third-party modules
from mongoengine.document import Document
from mongoengine.fields import StringField, IntField
import cachetools
id_lock = threading.Lock()
class Pool(Document):
meta = {
"collection": "noc.pools"
}
name = StringField(unique=True, min_length=1, max_length=16,
regex="^[0-9a-zA-Z]{1,16}$")
description = StringField()
discovery_reschedule_limit = IntField(default=50)
_id_cache = cachetools.TTLCache(1000, ttl=60)
_name_cache = cachetools.TTLCache(1000, ttl=60)
reschedule_lock = threading.Lock()
reschedule_ts = {} # pool id -> timestamp
def __unicode__(self):
return self.name
# -*- coding: utf-8 -*-
# ---------------------------------------------------------------------
# NewPrefixDiscoveryLog model
# ---------------------------------------------------------------------
# Copyright (C) 2007-2019 The NOC Project
# See LICENSE for details
# ---------------------------------------------------------------------
# Third-party modules
import six
from mongoengine.document import Document
from mongoengine.fields import StringField, DateTimeField
@six.python_2_unicode_compatible
class NewPrefixDiscoveryLog(Document):
meta = {
"collection": "noc.log.discovery.prefix.new",
"strict": False,
"auto_create_index": False,
"indexes": ["-timestamp"],
}
timestamp = DateTimeField()
vrf = StringField()
prefix = StringField()
description = StringField()
managed_object = StringField()
interface = StringField()
def __str__(self):
return "%s new %s:%s" % (self.timestamp, self.vrf, self.prefix)
use_tls = BooleanField()
def __str__(self):
return self.name or self.address
class AuthLDAPGroup(EmbeddedDocument):
# LDAP group dn
group_dn = StringField()
is_active = BooleanField()
group = ForeignKeyField(Group)
@on_save
@six.python_2_unicode_compatible
class AuthLDAPDomain(Document):
meta = {"collection": "noc.authldapdomain"}
name = StringField(unique=True)
is_active = BooleanField()
is_default = BooleanField()
description = StringField()
type = StringField(choices=[("ldap", "LDAP"), ("ad", "Active Directory")])
# Bind root
root = StringField()
# Users search tree
user_search_dn = StringField()
# Groups search tree
group_search_dn = StringField()
# Search expression to find user
# Use DEFAULT_USER_SEARCH_FILTER when empty
user_search_filter = StringField()
administrative_domain = ForeignKeyField(AdministrativeDomain)
selector = ForeignKeyField(ManagedObjectSelector)
time_pattern = ForeignKeyField(TimePattern)
min_severity = IntField(default=0)
# Action part
notification_group = ForeignKeyField(NotificationGroup)
template = ForeignKeyField(Template)
clear_template = ForeignKeyField(Template)
create_tt = BooleanField(default=False)
close_tt = BooleanField(default=False)
wait_tt = BooleanField(default=False)
# Stop or continue to next rule
stop_processing = BooleanField(default=False)
class AlarmEscalation(Document):
"""
Alarm escalations
"""
meta = {
"collection": "noc.alarmescalatons",
"allow_inheritance": False
}
name = StringField(unique=True)
description = StringField()
alarm_classes = ListField(EmbeddedDocumentField(AlarmClassItem))
pre_reasons = ListField(EmbeddedDocumentField(PreReasonItem))
escalations = ListField(EmbeddedDocumentField(EscalationItem))
global_limit = IntField()
def __unicode__(self):
from mongoengine.fields import StringField, IntField
import cachetools
# NOC modules
from noc.core.mongo.fields import ForeignKeyField, PlainReferenceField
from noc.main.models.style import Style
from noc.wf.models.workflow import Workflow
from noc.core.model.decorator import on_delete_check
from .phonenumberprofile import PhoneNumberProfile
id_lock = Lock()
@on_delete_check(check=[("phone.PhoneRange", "profile")])
@six.python_2_unicode_compatible
class PhoneRangeProfile(Document):
meta = {"collection": "noc.phonerangeprofiles", "strict": False, "auto_create_index": False}
name = StringField(unique=True)
description = StringField()
# Default phone number profile
default_number_profile = PlainReferenceField(PhoneNumberProfile)
# Cooldown time in days
# Time when number will be automatically transferred from C to F state
cooldown = IntField(default=30)
style = ForeignKeyField(Style)
workflow = PlainReferenceField(Workflow)
_id_cache = cachetools.TTLCache(100, ttl=60)
def __str__(self):
return self.name
query = PlainReferenceField(ConfDBQuery)
query_params = DictField()
filter_query = PlainReferenceField(ConfDBQuery)
is_active = BooleanField(default=True)
error_code = StringField()
error_text_template = StringField(default="{{error}}")
alarm_class = PlainReferenceField(AlarmClass)
is_fatal = BooleanField(default=False)
def __str__(self):
return self.query.name
@on_delete_check(check=[("inv.InterfaceProfile", "interface_validation_policy")])
@six.python_2_unicode_compatible
class InterfaceValidationPolicy(Document):
meta = {
"collection": "interfacevalidationpolicies",
"strict": False,
"auto_create_index": False,
}
name = StringField(unique=True)
description = StringField()
filter_query = PlainReferenceField(ConfDBQuery)
rules = ListField(EmbeddedDocumentField(InterfaceValidationRule))
_id_cache = cachetools.TTLCache(maxsize=100, ttl=60)
def __str__(self):
return self.name