mirror of
https://github.com/netbox-community/netbox.git
synced 2024-05-10 07:54:54 +00:00
391 lines
12 KiB
Python
391 lines
12 KiB
Python
import datetime
|
|
import json
|
|
import urllib
|
|
from collections import OrderedDict
|
|
from itertools import count, groupby
|
|
from typing import Any, Dict, List, Tuple
|
|
|
|
from django.core.serializers import serialize
|
|
from django.db.models import Count, OuterRef, Subquery
|
|
from django.db.models.functions import Coalesce
|
|
from jinja2.sandbox import SandboxedEnvironment
|
|
from mptt.models import MPTTModel
|
|
|
|
from dcim.choices import CableLengthUnitChoices
|
|
from extras.utils import is_taggable
|
|
from utilities.constants import HTTP_REQUEST_META_SAFE_COPY
|
|
|
|
|
|
def csv_format(data):
|
|
"""
|
|
Encapsulate any data which contains a comma within double quotes.
|
|
"""
|
|
csv = []
|
|
for value in data:
|
|
|
|
# Represent None or False with empty string
|
|
if value is None or value is False:
|
|
csv.append('')
|
|
continue
|
|
|
|
# Convert dates to ISO format
|
|
if isinstance(value, (datetime.date, datetime.datetime)):
|
|
value = value.isoformat()
|
|
|
|
# Force conversion to string first so we can check for any commas
|
|
if not isinstance(value, str):
|
|
value = '{}'.format(value)
|
|
|
|
# Double-quote the value if it contains a comma or line break
|
|
if ',' in value or '\n' in value:
|
|
value = value.replace('"', '""') # Escape double-quotes
|
|
csv.append('"{}"'.format(value))
|
|
else:
|
|
csv.append('{}'.format(value))
|
|
|
|
return ','.join(csv)
|
|
|
|
|
|
def foreground_color(bg_color, dark='000000', light='ffffff'):
|
|
"""
|
|
Return the ideal foreground color (dark or light) for a given background color in hexadecimal RGB format.
|
|
|
|
:param dark: RBG color code for dark text
|
|
:param light: RBG color code for light text
|
|
"""
|
|
bg_color = bg_color.strip('#')
|
|
r, g, b = [int(bg_color[c:c + 2], 16) for c in (0, 2, 4)]
|
|
if r * 0.299 + g * 0.587 + b * 0.114 > 186:
|
|
return dark
|
|
else:
|
|
return light
|
|
|
|
|
|
def dynamic_import(name):
|
|
"""
|
|
Dynamically import a class from an absolute path string
|
|
"""
|
|
components = name.split('.')
|
|
mod = __import__(components[0])
|
|
for comp in components[1:]:
|
|
mod = getattr(mod, comp)
|
|
return mod
|
|
|
|
|
|
def count_related(model, field):
|
|
"""
|
|
Return a Subquery suitable for annotating a child object count.
|
|
"""
|
|
subquery = Subquery(
|
|
model.objects.filter(
|
|
**{field: OuterRef('pk')}
|
|
).order_by().values(
|
|
field
|
|
).annotate(
|
|
c=Count('*')
|
|
).values('c')
|
|
)
|
|
|
|
return Coalesce(subquery, 0)
|
|
|
|
|
|
def serialize_object(obj, extra=None):
|
|
"""
|
|
Return a generic JSON representation of an object using Django's built-in serializer. (This is used for things like
|
|
change logging, not the REST API.) Optionally include a dictionary to supplement the object data. A list of keys
|
|
can be provided to exclude them from the returned dictionary. Private fields (prefaced with an underscore) are
|
|
implicitly excluded.
|
|
"""
|
|
json_str = serialize('json', [obj])
|
|
data = json.loads(json_str)[0]['fields']
|
|
|
|
# Exclude any MPTTModel fields
|
|
if issubclass(obj.__class__, MPTTModel):
|
|
for field in ['level', 'lft', 'rght', 'tree_id']:
|
|
data.pop(field)
|
|
|
|
# Include custom_field_data as "custom_fields"
|
|
if hasattr(obj, 'custom_field_data'):
|
|
data['custom_fields'] = data.pop('custom_field_data')
|
|
|
|
# Include any tags. Check for tags cached on the instance; fall back to using the manager.
|
|
if is_taggable(obj):
|
|
tags = getattr(obj, '_tags', None) or obj.tags.all()
|
|
data['tags'] = [tag.name for tag in tags]
|
|
|
|
# Append any extra data
|
|
if extra is not None:
|
|
data.update(extra)
|
|
|
|
# Copy keys to list to avoid 'dictionary changed size during iteration' exception
|
|
for key in list(data):
|
|
# Private fields shouldn't be logged in the object change
|
|
if isinstance(key, str) and key.startswith('_'):
|
|
data.pop(key)
|
|
|
|
return data
|
|
|
|
|
|
def dict_to_filter_params(d, prefix=''):
|
|
"""
|
|
Translate a dictionary of attributes to a nested set of parameters suitable for QuerySet filtering. For example:
|
|
|
|
{
|
|
"name": "Foo",
|
|
"rack": {
|
|
"facility_id": "R101"
|
|
}
|
|
}
|
|
|
|
Becomes:
|
|
|
|
{
|
|
"name": "Foo",
|
|
"rack__facility_id": "R101"
|
|
}
|
|
|
|
And can be employed as filter parameters:
|
|
|
|
Device.objects.filter(**dict_to_filter(attrs_dict))
|
|
"""
|
|
params = {}
|
|
for key, val in d.items():
|
|
k = prefix + key
|
|
if isinstance(val, dict):
|
|
params.update(dict_to_filter_params(val, k + '__'))
|
|
else:
|
|
params[k] = val
|
|
return params
|
|
|
|
|
|
def normalize_querydict(querydict):
|
|
"""
|
|
Convert a QueryDict to a normal, mutable dictionary, preserving list values. For example,
|
|
|
|
QueryDict('foo=1&bar=2&bar=3&baz=')
|
|
|
|
becomes:
|
|
|
|
{'foo': '1', 'bar': ['2', '3'], 'baz': ''}
|
|
|
|
This function is necessary because QueryDict does not provide any built-in mechanism which preserves multiple
|
|
values.
|
|
"""
|
|
return {
|
|
k: v if len(v) > 1 else v[0] for k, v in querydict.lists()
|
|
}
|
|
|
|
|
|
def deepmerge(original, new):
|
|
"""
|
|
Deep merge two dictionaries (new into original) and return a new dict
|
|
"""
|
|
merged = OrderedDict(original)
|
|
for key, val in new.items():
|
|
if key in original and isinstance(original[key], dict) and isinstance(val, dict):
|
|
merged[key] = deepmerge(original[key], val)
|
|
else:
|
|
merged[key] = val
|
|
return merged
|
|
|
|
|
|
def to_meters(length, unit):
|
|
"""
|
|
Convert the given length to meters.
|
|
"""
|
|
length = int(length)
|
|
if length < 0:
|
|
raise ValueError("Length must be a positive integer")
|
|
|
|
valid_units = CableLengthUnitChoices.values()
|
|
if unit not in valid_units:
|
|
raise ValueError(
|
|
"Unknown unit {}. Must be one of the following: {}".format(unit, ', '.join(valid_units))
|
|
)
|
|
|
|
if unit == CableLengthUnitChoices.UNIT_KILOMETER:
|
|
return length * 1000
|
|
if unit == CableLengthUnitChoices.UNIT_METER:
|
|
return length
|
|
if unit == CableLengthUnitChoices.UNIT_CENTIMETER:
|
|
return length / 100
|
|
if unit == CableLengthUnitChoices.UNIT_MILE:
|
|
return length * 1609.344
|
|
if unit == CableLengthUnitChoices.UNIT_FOOT:
|
|
return length * 0.3048
|
|
if unit == CableLengthUnitChoices.UNIT_INCH:
|
|
return length * 0.3048 * 12
|
|
raise ValueError(f"Unknown unit {unit}. Must be 'km', 'm', 'cm', 'mi', 'ft', or 'in'.")
|
|
|
|
|
|
def render_jinja2(template_code, context):
|
|
"""
|
|
Render a Jinja2 template with the provided context. Return the rendered content.
|
|
"""
|
|
return SandboxedEnvironment().from_string(source=template_code).render(**context)
|
|
|
|
|
|
def prepare_cloned_fields(instance):
|
|
"""
|
|
Compile an object's `clone_fields` list into a string of URL query parameters. Tags are automatically cloned where
|
|
applicable.
|
|
"""
|
|
params = []
|
|
for field_name in getattr(instance, 'clone_fields', []):
|
|
field = instance._meta.get_field(field_name)
|
|
field_value = field.value_from_object(instance)
|
|
|
|
# Pass False as null for boolean fields
|
|
if field_value is False:
|
|
params.append((field_name, ''))
|
|
|
|
# Omit empty values
|
|
elif field_value not in (None, ''):
|
|
params.append((field_name, field_value))
|
|
|
|
# Copy tags
|
|
if is_taggable(instance):
|
|
for tag in instance.tags.all():
|
|
params.append(('tags', tag.pk))
|
|
|
|
# Concatenate parameters into a URL query string
|
|
param_string = '&'.join([f'{k}={v}' for k, v in params])
|
|
|
|
return param_string
|
|
|
|
|
|
def shallow_compare_dict(source_dict, destination_dict, exclude=None):
|
|
"""
|
|
Return a new dictionary of the different keys. The values of `destination_dict` are returned. Only the equality of
|
|
the first layer of keys/values is checked. `exclude` is a list or tuple of keys to be ignored.
|
|
"""
|
|
difference = {}
|
|
|
|
for key in destination_dict:
|
|
if source_dict.get(key) != destination_dict[key]:
|
|
if isinstance(exclude, (list, tuple)) and key in exclude:
|
|
continue
|
|
difference[key] = destination_dict[key]
|
|
|
|
return difference
|
|
|
|
|
|
def flatten_dict(d, prefix='', separator='.'):
|
|
"""
|
|
Flatten netsted dictionaries into a single level by joining key names with a separator.
|
|
|
|
:param d: The dictionary to be flattened
|
|
:param prefix: Initial prefix (if any)
|
|
:param separator: The character to use when concatenating key names
|
|
"""
|
|
ret = {}
|
|
for k, v in d.items():
|
|
key = separator.join([prefix, k]) if prefix else k
|
|
if type(v) is dict:
|
|
ret.update(flatten_dict(v, prefix=key))
|
|
else:
|
|
ret[key] = v
|
|
return ret
|
|
|
|
|
|
def decode_dict(encoded_dict: Dict, *, decode_keys: bool = True) -> Dict:
|
|
"""
|
|
Recursively URL decode string keys and values of a dict.
|
|
|
|
For example, `{'1%2F1%2F1': {'1%2F1%2F2': ['1%2F1%2F3', '1%2F1%2F4']}}` would
|
|
become: `{'1/1/1': {'1/1/2': ['1/1/3', '1/1/4']}}`
|
|
|
|
:param encoded_dict: Dictionary to be decoded.
|
|
:param decode_keys: (Optional) Enable/disable decoding of dict keys.
|
|
"""
|
|
|
|
def decode_value(value: Any, _decode_keys: bool) -> Any:
|
|
"""
|
|
Handle URL decoding of any supported value type.
|
|
"""
|
|
# Decode string values.
|
|
if isinstance(value, str):
|
|
return urllib.parse.unquote(value)
|
|
# Recursively decode each list item.
|
|
elif isinstance(value, list):
|
|
return [decode_value(v, _decode_keys) for v in value]
|
|
# Recursively decode each tuple item.
|
|
elif isinstance(value, Tuple):
|
|
return tuple(decode_value(v, _decode_keys) for v in value)
|
|
# Recursively decode each dict key/value pair.
|
|
elif isinstance(value, dict):
|
|
# Don't decode keys, if `decode_keys` is false.
|
|
if not _decode_keys:
|
|
return {k: decode_value(v, _decode_keys) for k, v in value.items()}
|
|
return {urllib.parse.unquote(k): decode_value(v, _decode_keys) for k, v in value.items()}
|
|
return value
|
|
|
|
if not decode_keys:
|
|
# Don't decode keys, if `decode_keys` is false.
|
|
return {k: decode_value(v, decode_keys) for k, v in encoded_dict.items()}
|
|
|
|
return {urllib.parse.unquote(k): decode_value(v, decode_keys) for k, v in encoded_dict.items()}
|
|
|
|
|
|
# Taken from django.utils.functional (<3.0)
|
|
def curry(_curried_func, *args, **kwargs):
|
|
def _curried(*moreargs, **morekwargs):
|
|
return _curried_func(*args, *moreargs, **{**kwargs, **morekwargs})
|
|
return _curried
|
|
|
|
|
|
def array_to_string(array):
|
|
"""
|
|
Generate an efficient, human-friendly string from a set of integers. Intended for use with ArrayField.
|
|
For example:
|
|
[0, 1, 2, 10, 14, 15, 16] => "0-2, 10, 14-16"
|
|
"""
|
|
group = (list(x) for _, x in groupby(sorted(array), lambda x, c=count(): next(c) - x))
|
|
return ', '.join('-'.join(map(str, (g[0], g[-1])[:len(g)])) for g in group)
|
|
|
|
|
|
def content_type_name(contenttype):
|
|
"""
|
|
Return a proper ContentType name.
|
|
"""
|
|
try:
|
|
meta = contenttype.model_class()._meta
|
|
return f'{meta.app_config.verbose_name} > {meta.verbose_name}'
|
|
except AttributeError:
|
|
# Model no longer exists
|
|
return f'{contenttype.app_label} > {contenttype.model}'
|
|
|
|
|
|
#
|
|
# Fake request object
|
|
#
|
|
|
|
class NetBoxFakeRequest:
|
|
"""
|
|
A fake request object which is explicitly defined at the module level so it is able to be pickled. It simply
|
|
takes what is passed to it as kwargs on init and sets them as instance variables.
|
|
"""
|
|
def __init__(self, _dict):
|
|
self.__dict__ = _dict
|
|
|
|
|
|
def copy_safe_request(request):
|
|
"""
|
|
Copy selected attributes from a request object into a new fake request object. This is needed in places where
|
|
thread safe pickling of the useful request data is needed.
|
|
"""
|
|
meta = {
|
|
k: request.META[k]
|
|
for k in HTTP_REQUEST_META_SAFE_COPY
|
|
if k in request.META and isinstance(request.META[k], str)
|
|
}
|
|
return NetBoxFakeRequest({
|
|
'META': meta,
|
|
'POST': request.POST,
|
|
'GET': request.GET,
|
|
'FILES': request.FILES,
|
|
'user': request.user,
|
|
'path': request.path,
|
|
'id': getattr(request, 'id', None), # UUID assigned by middleware
|
|
})
|