#!/usr/bin/env python3
# -*- coding: utf-8 -*-
# -------------------------------------------------------------------------------
# This file is part of Mentat system (https://mentat.cesnet.cz/).
#
# Copyright (C) since 2011 CESNET, z.s.p.o (http://www.ces.net/)
# Use of this source is governed by the MIT license, see LICENSE file.
# -------------------------------------------------------------------------------
"""
This module contains various usefull utilities for *Hawat* application.
"""
__author__ = "Jan Mach <jan.mach@cesnet.cz>"
__credits__ = "Pavel Kácha <pavel.kacha@cesnet.cz>, Andrea Kropáčová <andrea.kropacova@cesnet.cz>"
import os
import uuid
import copy
import datetime
import json
import yaml
[docs]class URLParamsBuilder:
"""
Small utility class for building URL parameter dictionaries for various view
endpoints.
.. note::
This class is still proof of concept and work in progress.
"""
def __init__(self, skeleton = None):
self.rules = []
self.kwrules = {}
self.skeleton = skeleton or {}
@staticmethod
def _add_scalar(dst, key, val):
if val is not None:
dst[key] = val
@staticmethod
def _add_vector(dst, key, val):
if val is not None:
dst.setdefault(key, []).append(val)
[docs] def add_rule(self, key, as_list = False, optional = False):
"""
Add new rule to URL parameter builder.
:param str key: Name of the rule key.
:param bool as_list: Indication that the rule parameter is a list of multiple values.
:param bool optional: Indication that the rule parameter is optional.
"""
if as_list:
rule = [key, self._add_vector, True, optional]
self.rules.append(rule)
else:
rule = [key, self._add_scalar, False, optional]
self.rules.append(rule)
return self
[docs] def add_kwrule(self, key, as_list = False, optional = False):
"""
Add new keyword rule to URL parameter builder.
:param str key: Name of the rule key.
:param bool as_list: Indication that the rule parameter is a list of multiple values.
:param bool optional: Indication that the rule parameter is optional.
"""
if as_list:
rule = [key, self._add_vector, True, optional]
self.kwrules[key] = rule
else:
rule = [key, self._add_scalar, False, optional]
self.kwrules[key] = rule
return self
[docs] def get_params(self, *args, **kwargs):
"""
Get URL parameters as dictionary with filled-in values.
"""
tmp = copy.deepcopy(self.skeleton)
for idx, rule in enumerate(self.rules):
try:
rule[1](tmp, rule[0], args[idx])
except IndexError:
if not rule[3]:
raise
for key, rule in self.kwrules.items():
if key in kwargs:
rule[1](tmp, rule[0], kwargs[key])
return tmp
[docs]class LimitCounter:
"""
Simple configurable limit counter with support for multiple keys.
"""
def __init__(self, limit):
self.counters = {}
self.limit = limit
[docs] def count_and_check(self, key, increment = 1):
"""
Increment key counter and check against internal limit.
"""
self.counters[key] = self.counters.get(key, 0) + increment
return self.counters[key] <= self.limit
#------------------------------------------------------------------------------
[docs]def get_timedelta(tstamp):
"""
Get timedelta from current UTC time and given datetime object.
:param datetime.datetime: Datetime of the lower timedelta boundary.
:return: Timedelta object.
:rtype: datetime.timedelta
"""
return datetime.datetime.utcnow() - tstamp
[docs]def get_datetime_utc(aware = False):
"""
Get current UTC datetime.
:return: Curent UTC datetime.
:rtype: datetime.datetime
"""
if aware:
return datetime.datetime.now(datetime.timezone.utc)
return datetime.datetime.utcnow()
[docs]def parse_datetime(dtstring):
"""
Parse given datetime string.
:param str dtstring: Datetime string in ISON format to parse.
:return: Curent UTC datetime.
:rtype: datetime.datetime
"""
return datetime.datetime.fromisoformat(dtstring)
[docs]def get_datetime_local():
"""
Get current local timestamp.
:return: Curent local timestamp.
:rtype: datetime.datetime
"""
return datetime.datetime.now()
[docs]def check_file_exists(filename):
"""
Check, that given file exists in the filesystem.
:param str filename: Name of the file to check.
:return: Existence flag as ``True`` or ``False``.
:rtype: bool
"""
return os.path.isfile(filename)
[docs]def in_query_params(haystack, needles, on_true = True, on_false = False, on_empty = False):
"""
Utility method for checking that any needle from given list of needles is
present in given haystack.
"""
if not haystack:
return on_empty
for needle in needles:
if needle in haystack:
return on_true
return on_false
[docs]def generate_query_params(baseparams, updates):
"""
Generate query parameters for GET method form.
:param dict baseparams: Original query parameters.
:param dict updates: Updates for query parameters.
:return: Deep copy of original parameters modified with given updates.
:rtype: dict
"""
result = copy.deepcopy(baseparams)
result.update(updates)
return result
[docs]def json_to_yaml(json_data):
"""
Include given file in raw form directly into the generated content.
This may be usefull for example for including JavaScript files
directly into the HTML page.
"""
return yaml.dump(
yaml.safe_load(
json_data
),
default_flow_style=False
)
[docs]def get_uuid4():
"""
Generate random UUID identifier.
"""
return uuid.uuid4()
[docs]def load_json_from_file(filename):
"""
Load JSON from given file.
"""
with open(filename, encoding="utf8") as fhnd:
res = json.load(fhnd)
return res
[docs]def make_copy_deep(data):
"""
Make a deep copy of given data structure.
"""
return copy.deepcopy(data)