def scheduler(cls,
interval: (crontab, float),
*args,
queue: Enum = TaskQueue.SHORT,
**kwargs): # pragma: no cover
"""
Registers the decorated function as a periodic task. The task should
not accept any arguments.
:param interval: Periodic interval in seconds as float or crontab
object specifying task trigger time. See
http://docs.celeryproject.org/en/latest/reference/celery.schedules.html#celery.schedules.crontab
:param queue: Queue to use for the scheduled task.
:param args: Arguments to pass to scheduled task.
:param kwargs: Keyword arguments to pass to scheduled task.
"""
def _wrapper(function: Callable):
task = celery.task(function,
base=ExceptionLoggerTask,
queue=queue.value)
celery.add_periodic_task(interval, task.s(), args, kwargs)
return function
return _wrapper
python类Enum()的实例源码
def __init__(self, parameter, position, type_=None):
default = parameter.default
empty = parameter.empty
self._parameter = parameter
self.is_positional = default is empty
self.is_optional = not self.is_positional
self.is_keyword_only = self.kind is parameter.KEYWORD_ONLY
if type_ is not None:
self.type = type_
self.is_bool = issubclass(type_, bool)
self.is_dict = issubclass(type_, dict)
self.is_enum = issubclass(type_, Enum)
self.is_list = issubclass(type_, (list, tuple))
else:
self.type = str if default in (None, empty) else type(default)
self.is_bool = isinstance(default, bool)
self.is_dict = isinstance(default, dict)
self.is_enum = isinstance(default, Enum)
self.is_list = isinstance(default, (list, tuple))
self.position = position
self.takes_value = self.is_positional or (self.is_optional and not self.is_bool)
def __getattribute__(self, name):
# Intercept everything and delegate to the device class by converting
# names between the D-Bus conventions to Python conventions.
prop_name = camel_to_snake(name)
if (prop_name in DeviceAPI._PROPERTIES or prop_name in DeviceAPI._RW_PROPERTIES) \
and hasattr(self._driver, prop_name):
value = getattr(self._driver, prop_name)
if isinstance(value, Enum):
return value.name.lower()
if isinstance(value, Color):
return value.html
if isinstance(value, (list, tuple)) and len(value) > 0 and isinstance(value[0], Enum):
return [x.name.lower() for x in value]
return value
else:
return super(DeviceAPI, self).__getattribute__(name)
def __init__(self, **kwargs):
""" U-Boot Image Header Constructor
:param laddr: Load address
:param eaddr: Entry point address
:param arch: Architecture (ARCHType Enum)
:param os: Operating system (OSType Enum)
:param image: Image type (IMGType Enum)
:param compress: Image compression (COMPRESSType Enum)
:param name: Image name (max: 32 chars)
"""
self.MagicNumber = 0x27051956 # U-Boot Default Value is 0x27051956
self.TimeStamp = int(time.time())
self.DataSize = 0
self.DataCRC = 0
self.LoadAddress = 0 if 'laddr' not in kwargs else kwargs['laddr']
self.EntryAddress = 0 if 'eaddr' not in kwargs else kwargs['eaddr']
self.OsType = OSType.LINUX if 'os' not in kwargs else kwargs['os']
self.ArchType = ARCHType.ARM if 'arch' not in kwargs else kwargs['arch']
self.ImageType = IMGType.STD if 'image' not in kwargs else kwargs['image']
self.Compression = COMPRESSType.NONE if 'compress' not in kwargs else kwargs['compress']
self.Name = '' if 'name' not in kwargs else kwargs['name']
def __init__(self):
self.entities = {
'<cuisine>' : None,
'<location>' : None,
'<party_size>' : None,
'<rest_type>' : None,
}
self.num_features = 4 # tracking 4 entities
self.rating = None
# constants
self.party_sizes = ['??', '??', '??', '?', '??', '?', '??', '???', '??', '???', '??', '??']
self.locations = ['??', '???', '???', '???', '??', '??', '??', '????', '??', '??', 'LA']
self.cuisines = ['??','??','???', '????', '??', '??', '??', '??', '??', '???', '??', '???']
self.rest_types = ['??', '?','??', '??']
self.EntType = Enum('Entity Type', '<party_size> <location> <cuisine> <rest_type> <non_ent>')
def test_without_values(self):
self.assertGeneratedOutput(
"""
enum Bar {
TOP,
RIGHT,
BOTTOM,
LEFT
};
""",
"""
from enum import Enum
class Bar(Enum):
TOP = 0
RIGHT = 1
BOTTOM = 2
LEFT = 3
"""
)
def test_values(self):
self.assertGeneratedOutput(
"""
enum Bar {
TOP = 37,
RIGHT = 42,
BOTTOM = 55,
LEFT = 69
};
""",
"""
from enum import Enum
class Bar(Enum):
TOP = 37
RIGHT = 42
BOTTOM = 55
LEFT = 69
"""
)
def test_initial_values(self):
self.assertGeneratedOutput(
"""
enum Bar {
TOP = 37,
RIGHT,
BOTTOM,
LEFT
};
""",
"""
from enum import Enum
class Bar(Enum):
TOP = 37
RIGHT = 38
BOTTOM = 39
LEFT = 40
"""
)
def test_expressions_for_values(self):
self.assertGeneratedOutput(
"""
enum Bar {
TOP = 1 << 0,
RIGHT = 1 << 1,
BOTTOM = 1 << 2,
LEFT = 1 << 3
};
""",
"""
from enum import Enum
class Bar(Enum):
TOP = 1
RIGHT = 2
BOTTOM = 4
LEFT = 8
"""
)
def test_inner_enum(self):
self.assertGeneratedOutput(
"""
class Point {
enum Temperature {
HOT,
COLD
};
};
""",
"""
from enum import Enum
class Point:
class Temperature(Enum):
HOT = 0
COLD = 1
"""
)
def __get_ie(ie):
if ie is None:
imp = exp = lambda val: val
elif ie is NotImplemented:
imp = exp = lambda val: NotImplemented
elif isinstance(ie, basestring):
imp, exp = _object_proxy(ie), lambda val: val.id
elif isinstance(ie, types.ModuleType):
imp, exp = ie.loads, ie.dumps
elif isinstance(ie, type) and issubclass(ie, BaseObject):
imp, exp = ie, lambda val: val.id
elif isinstance(ie, type) and issubclass(ie, Enum):
imp, exp = ie, lambda val: val.value
elif callable(ie):
imp = exp = ie
return imp, exp
def __get_search(search, ie):
if search is False:
return None
if search is True:
if isinstance(ie, tuple):
ie = ie[1]
return __get_search(ie, False)
elif search is None:
return lambda val: val
elif isinstance(search, basestring):
if search == 'User':
return lambda val: val.passportname
elif search == 'FormattedText':
return lambda val: val.raw
else:
return lambda val: val.name
elif isinstance(search, type) and issubclass(search, Enum):
return lambda val: val.format_name
elif callable(search):
return search
return None
def default(self, o):
# for Enum Type
if isinstance(o, enum.Enum):
return o.value
# for Enum Select Integer
if isinstance(o, EnumInt):
return o.key
if isinstance(o, (datetime, date)):
return o.isoformat()
if isinstance(o, Decimal):
return _number_str(o)
if isinstance(o, ObjectId):
return str(o)
return super(JSONEncoder, self).default(o)
def _datetimeholders_compare(self, a: DateTimeHolder, b: DateTimeHolder, from_fraction: Enum):
"""Partially compare a and b date time holders, starting with fraction.
For example, if the fraction is DAY, compare only DAY, MONTH and YEAR
"""
_a = DateTimeHolder()
_b = DateTimeHolder()
for fraction_value in range(from_fraction.value, self.highest_fraction.value+1):
fraction = self.fractions(fraction_value)
_a[fraction.name] = a[fraction.name]
_b[fraction.name] = b[fraction.name]
if _a > _b:
return 1
elif _a == _b:
return 0
else:
return -1
def _send_command(self, name, items):
for key, value in items.items():
if isinstance(value, Enum):
items[key] = value.value
if "bd_addr" in items:
items["bd_addr"] = FlicClient._bdaddr_string_to_bytes(items["bd_addr"])
opcode = FlicClient._COMMAND_NAME_TO_OPCODE[name]
data_bytes = FlicClient._COMMAND_STRUCTS[opcode].pack(*FlicClient._COMMAND_NAMED_TUPLES[opcode](**items))
bytes = bytearray(3)
bytes[0] = (len(data_bytes) + 1) & 0xff
bytes[1] = (len(data_bytes) + 1) >> 8
bytes[2] = opcode
bytes += data_bytes
with self._lock:
if not self._closed:
self._sock.sendall(bytes)
def setUp(self):
import enum, ctypes
class Fruit(enum.Enum):
Apple, Banana, Citron = range(3)
class TestStruct(PokeStructure):
_fields_ = [
("_fruit", ctypes.c_uint8),
]
_adapters_ = [
("_fruit", Fruit)
]
self.Fruit = Fruit
self.TestStruct = TestStruct
def adapt(self, impltype, **kw):
schema = kw.pop('schema', self.schema)
metadata = kw.pop('metadata', self.metadata)
_create_events = kw.pop('_create_events', False)
if issubclass(impltype, Enum):
if self.enum_class is not None:
args = [self.enum_class]
else:
args = self.enums
return impltype(name=self.name,
schema=schema,
metadata=metadata,
convert_unicode=self.convert_unicode,
native_enum=self.native_enum,
inherit_schema=self.inherit_schema,
validate_strings=self.validate_strings,
_create_events=_create_events,
*args,
**kw)
else:
# TODO: why would we be here?
return super(Enum, self).adapt(impltype, **kw)
def before_write_item(self, value):
source_value = value.name if isinstance(value, Enum) else value
enum_cls = self.enum_cls
# Check real enum value
try:
if isinstance(source_value, compat.string_types):
return enum_cls[source_value].value
else:
return enum_cls(source_value).value
except (ValueError, KeyError):
choices = ', '.join(
"'{}' = {}".format(x.name, x.value) for x in enum_cls
)
enum_str = '{}({})'.format(enum_cls.__name__, choices)
raise errors.LogicalError(
"Unknown element '{}' for type {}"
.format(source_value, enum_str)
)
def create_enum_column(spec):
if spec.startswith('Enum8'):
params = spec[6:-1]
cls = Enum8Column
else:
params = spec[7:-1]
cls = Enum16Column
d = {}
for param in params.split(", '"):
pos = param.rfind("'")
name = param[:pos].lstrip("'")
value = int(param[pos + 1:].lstrip(' ='))
d[name] = value
return cls(Enum(cls.ch_type, d))
def escape_param(item):
if item is None:
return 'NULL'
elif isinstance(item, datetime):
return "'%s'" % item.strftime('%Y-%m-%d %H:%M:%S')
elif isinstance(item, date):
return "'%s'" % item.strftime('%Y-%m-%d')
elif isinstance(item, string_types):
return "'%s'" % ''.join(escape_chars_map.get(c, c) for c in item)
elif isinstance(item, (list, tuple)):
return "[%s]" % ', '.join(text_type(escape_param(x)) for x in item)
elif isinstance(item, Enum):
return item.value
elif isinstance(item, UUID):
return "'%s'" % str(item)
else:
return item
def resolve_topic_name(topic):
"""
Given a topic of type `enum.Enum`, or of type `str`,
get the string value of its publisher channel.
Args:
topic (enum.Enum | str): the topic to resolve, as
either an enum.Enum type or a string.
Returns:
string: the publisher channel for the topic
None: if the topic arg is not one of the two acceptable
types, then None is returned
"""
if isinstance(topic, enum.Enum):
return topic.value
elif isinstance(topic, str):
return topic
else:
return None
def adapt(self, impltype, **kw):
schema = kw.pop('schema', self.schema)
metadata = kw.pop('metadata', self.metadata)
_create_events = kw.pop('_create_events', False)
if issubclass(impltype, Enum):
if self.enum_class is not None:
args = [self.enum_class]
else:
args = self.enums
return impltype(name=self.name,
schema=schema,
metadata=metadata,
convert_unicode=self.convert_unicode,
native_enum=self.native_enum,
inherit_schema=self.inherit_schema,
validate_strings=self.validate_strings,
_create_events=_create_events,
*args,
**kw)
else:
# TODO: why would we be here?
return super(Enum, self).adapt(impltype, **kw)
def responder(cls, plugin_name: str, *actions: [Enum],
queue: Enum=TaskQueue.SHORT):
"""
Registers the decorated function as a responder to the actions
provided. Specifying description as defaults on option specific args
is mandatory.
"""
def _wrapper(function):
task = celery.task(function,
base=ExceptionLoggerTask,
queue=queue.value)
for action in actions:
cls._responders[action].append(task)
cls._plugins[task] = plugin_name
params = signature(function).parameters.values()
cls._options[task] = [param.name for param in params
if param.default is not Parameter.empty]
return function
return _wrapper
def add_subscriber(self, obj, name, entity):
"""
:param object obj: Object to be called when the event fires
:param str|Enum name: Name of the event. May be a string-valued enum.
:param entity: If ``None``, receive all events regardless of their entity.
Otherwise, only receive events whose entity ``is`` this
object.
Store ``(obj, entity)`` as a subscriber for the event *name*. When
:py:meth:`fire` is called with a pattern that matches ``(obj, entity)``,
call ``obj.on_[name](Event(name, entity, data))``.
An event is said to "match" a subscription if the subscription has the same
event name, and either the subscriber's entity is ``None``, or the
subscriber's entity ``is`` the event's entity.
You may subscribe more than once to receive the event multiple times.
"""
if isinstance(name, Enum):
name = name.value
self.handlers[name].append((obj, entity))
def fire(self, name, entity, data):
"""
:param object obj: Object that is subscribed
:param str|Enum name: Name of the event. May be a string-valued enum.
:param entity: Entity, or ``None``.
:param data: Arbitrary data to add to the :py:class:`Event`.
Call all event handlers for the given *name* + *entity*, and pass *data*.
"""
if isinstance(name, Enum):
name = name.value
method_name = "on_" + name.lower()
event = Event(name, entity, data)
for (obj, required_entity) in self.handlers[name]:
if required_entity is None or entity is required_entity:
method = getattr(obj, method_name)
method(event)
if event._is_halted or self._is_halted:
break
self._is_halted = False
def _format_enum_values(values: List[int],
enum_class: Type[Enum]):
"""
Formats a list of enum values represented by their numeric value, which may contain missing values indicated by
None.
Returns a list containing a string representation for each list element, which is "None" if the element is None, or
the name of the enum member corresponding to the numeric value in the list otherwise.
Parameters
----------
values: list
A list of numeric enum values
enum_class: Type[Enum]
The enum type
Returns
-------
list
A list of strings containing a string representation for each element in the specified list
"""
return ["None" if x == _MISSING_VALUE_INT else enum_class(x).name for x in values]
def __init__(self,canvas):
self.isOpen = False
self.SegType = Enum("SegType", "Straight Biarc End")
self.point = [ coords(0,0) ]
self.tangent = [None]
self.biarc_r = []
self.segwidth = []
self.segtype = []
self.cidmap = {}
self.color = "gray"
self.tag = "segment"
self.movePoint(0,coords(100,100))
self.appendPoint(coords(200,110),self.SegType.Straight,None,30)
#self.appendPoint(coords(320,310),self.SegType.Biarc,30,1)
self.appendPoint(coords(520,400),self.SegType.Biarc,None,30,1)
self.insertPoint(2,coords(320,310),self.SegType.Biarc,None,30,1)
self.insertPoint(2,coords(620,610),self.SegType.Straight,None,30,1)
self.removePoint(2)
def json_encode(obj):
if( isinstance(obj,(Device,DeviceType,Attribute,Parameter))):
return {key.lstrip('_'): value for key, value in obj.__dict__.items()}
if(isinstance(obj,Enum)):
return str(obj.value)
if( isinstance(obj,bytes)):
return str(UUID(bytes=obj))
if( isinstance(obj,UUID)):
return str(obj)
if (isinstance(obj, type(threading.Lock()))):
return str(obj)
if (isinstance(obj, threading.Condition)):
return str(obj)
if (isinstance(obj, threading.Thread)):
return str(obj)
else:
return
def todict(obj): # pylint: disable=too-many-return-statements
if isinstance(obj, dict):
return {k: todict(v) for (k, v) in obj.items()}
elif isinstance(obj, list):
return [todict(a) for a in obj]
elif isinstance(obj, Enum):
return obj.value
elif isinstance(obj, (date, time, datetime)):
return obj.isoformat()
elif isinstance(obj, timedelta):
return str(obj)
elif hasattr(obj, '_asdict'):
return todict(obj._asdict())
elif hasattr(obj, '__dict__'):
return dict([(to_camel_case(k), todict(v))
for k, v in obj.__dict__.items()
if not callable(v) and not k.startswith('_')])
return obj
def test_querying_table(metadata):
"""
Create an object for test table.
"""
# When using pytest-xdist, we don't want concurrent table creations
# across test processes so we assign a unique name for table based on
# the current worker id.
worker_id = os.environ.get('PYTEST_XDIST_WORKER', 'master')
return Table(
'test_querying_table_' + worker_id, metadata,
Column('id', types.Integer, autoincrement=True, primary_key=True),
Column('t_string', types.String(60)),
Column('t_list', types.ARRAY(types.String(60))),
Column('t_enum', types.Enum(MyEnum)),
Column('t_int_enum', types.Enum(MyIntEnum)),
Column('t_datetime', types.DateTime()),
Column('t_date', types.DateTime()),
Column('t_interval', types.Interval()),
Column('uniq_uuid', PG_UUID, nullable=False, unique=True, default=uuid4),
)
