#
# Utilities for filtering on leases
+# w0622: we keep on using 'filter' as a variable name
+# pylint: disable=c0111, c0103, w0622
+
import time
-import calendar
-from PLC.Faults import *
+from PLC.Faults import * # pylint: disable=w0401, w0614
from PLC.Filter import Filter
from PLC.Parameter import Parameter, Mixed
from PLC.Timestamp import Timestamp
# supersede the generic Filter class to support time intersection
-class LeaseFilter (Filter):
+class LeaseFilter(Filter):
# general notes on input parameters
# int_timestamp: number of seconds since the epoch
# str_timestamp: see Timestamp.sql_validate
- # timeslot: a tuple (from, until), each being either int_timestamp or
+ # timeslot: a list [from, until], each being either int_timestamp or
# str_timestamp
local_fields = {
'alive': Mixed(
- Parameter(
- int, "int_timestamp: leases alive at that time"),
- Parameter(
- str, "str_timestamp: leases alive at that time"),
- Parameter(
- tuple, "timeslot: the leases alive during this timeslot")),
- 'clip': Mixed(
- Parameter(
- int, "int_timestamp: leases alive after that time"),
- Parameter(
- str, "str_timestamp: leases alive after at that time"),
- Parameter(
- tuple, "timeslot: the leases alive during this timeslot")),
+ Parameter(int, "int_timestamp: leases alive at that time"),
+ Parameter(str, "str_timestamp: leases alive at that time"),
+ Parameter(list, "timeslot: the leases alive during this timeslot"),
+ ),
+ 'clip': Mixed(
+ Parameter(int, "int_timestamp: leases alive after that time"),
+ Parameter(str, "str_timestamp: leases alive after that time"),
+ Parameter(list, "timeslot: the leases alive during this timeslot"),
+ ),
########## macros
# {'day' : 0} : all leases from today and on
# {'day' : 1} : all leases today (localtime at the myplc)
# {'day' : 2} : all leases today and tomorrow (localtime at the myplc)
# etc..
- 'day': Parameter(int, "clip on a number of days from today and on;"
- " 0 means no limit in the future"),
+ 'day': Parameter(
+ int,
+ "clip on a number of days from today and on;"
+ " 0 means no limit in the future"),
}
- def __init__(self, fields={}, filter={},
+ def __init__(self, fields=None, filter=None,
doc="Lease filter -- adds the 'alive' and 'clip'"
"capabilities for filtering on leases"):
+ if fields is None:
+ fields = {}
+ if filter is None:
+ filter = {}
Filter.__init__(self, fields, filter, doc)
self.fields.update(LeaseFilter.local_fields)
# canonical type
@staticmethod
- def quote(timestamp): return Timestamp.cast_long(timestamp)
+ def quote(timestamp):
+ return Timestamp.cast_long(timestamp)
# basic SQL utilities
@staticmethod
def sql_time_intersect(f1, u1, f2, u2):
- # either f2 is in [f1,u1], or u2 is in [f1,u1], or f2<=f1<=u1<=u2
- return ("(({f1} <= {f2}) AND ({f2} <= {u1})) " +
- "OR (({f1} <= {u2}) AND ({u2} <= {u1})) " +
- "OR (({f2}<={f1}) AND ({u1}<={u2}))").format(**locals())
+ # either f2 is in [f1,u1[, or u2 is in ]f1,u1], or f2<=f1<u1<=u2
+ return (
+ f"(({f1} <= {f2}) AND ({f2} < {u1})) "
+ f"OR (({f1} < {u2}) AND ({u2} <= {u1})) "
+ f"OR (({f2}<={f1}) AND ({u1}<={u2}))")
@staticmethod
def time_in_range(timestamp, f1, u1):
@staticmethod
def sql_time_in_range(timestamp, f1, u1):
# is timestamp in [f1, u1]
- return "(({f1} <= {timestamp}) AND ({timestamp} <= {u1}))"\
- .format(**locals())
+ return f"(({f1} <= {timestamp}) AND ({timestamp} <= {u1}))"
@staticmethod
def sql_timeslot_after(f1, u1, mark):
# is the lease alive after mark, i.e. u1 >= mark
- return "({u1} >= {mark})".format(**locals())
+ return f"({u1} >= {mark})"
# hooks for the local fields
def sql_alive(self, alive):
- if isinstance(alive, int) or isinstance(alive, str):
+ if isinstance(alive, (int, str)):
# the lease is alive at that time if from <= alive <= until
alive = LeaseFilter.quote(alive)
return LeaseFilter.sql_time_in_range(alive, 't_from', 't_until')
- elif isinstance(alive, tuple):
+ elif isinstance(alive, (tuple, list)):
(f, u) = alive
f = LeaseFilter.quote(f)
u = LeaseFilter.quote(u)
return LeaseFilter.sql_time_intersect(f, u, 't_from', 't_until')
else:
- raise PLCInvalidArgument("LeaseFilter: alive field {}"
- .format(alive))
+ raise PLCInvalidArgument(f"LeaseFilter: alive field {alive}")
def sql_clip(self, clip):
if isinstance(clip, int) or isinstance(clip, str):
start = LeaseFilter.quote(clip)
return LeaseFilter.sql_timeslot_after('t_from', 't_until', start)
- elif isinstance(clip, tuple):
+ elif isinstance(clip, (tuple, list)):
(f, u) = clip
f = LeaseFilter.quote(f)
u = LeaseFilter.quote(u)
return LeaseFilter.sql_time_intersect(f, u, 't_from', 't_until')
else:
- raise PLCInvalidArgument("LeaseFilter: clip field {}"
- .format(clip))
+ raise PLCInvalidArgument(f"LeaseFilter: clip field {clip}")
# the whole key to implementing day is to compute today's beginning
def today_start(self):
try:
# locate hook function associated with key
method = LeaseFilter.__dict__['sql_' + k]
- where_part += " {} {}({})"\
- .format(self.join_with,
- self.negation[k],
- method(self, self.local[k]))
+ where_part += (
+ f" {self.join_with} {self.negation[k]}"
+ f"({method(self, self.local[k])})"
+ )
except Exception as e:
raise PLCInvalidArgument(
- "LeaseFilter: something wrong with filter"
- "key {}, val was {} -- {}".format(k, v, e))
+ f"LeaseFilter: something wrong with filter "
+ f"key {k}, value was {v} -- {e}")
return (where_part, clip_part)
# xxx not sure where this belongs yet
git://git.onelab.eu / plcapi.git / blobdiff