How to use _to_datetime method in pandera

Best Python code snippet using pandera_python

celmon.py

Source:celmon.py

...3from datetime import datetime as dt4class Logger:5    def log_task_status_change(self, task, event):6        print('[{}] {} {} (STATE={}, UUID={})'.format(7            self._to_datetime(task.timestamp),8            event['type'].upper(),9            task.name,10            task.state.upper(),11            task.uuid12        ))13    def log_event_details(self, event):14        print('EVENT DETAILS: {}'.format(event))15    def log_task_details(self, task):16        print('TASK DETAILS:')17        print('UUID: {}'.format(task.uuid))18        print('Name: {}'.format(task.name))19        print('State: {}'.format(task.state))20        print('Received: {}'.format(self._to_datetime(task.received)))21        print('Sent: {}'.format(self._to_datetime(task.sent)))22        print('Started: {}'.format(self._to_datetime(task.started)))23        print('Rejected: {}'.format(self._to_datetime(task.rejected)))24        print('Succeeded: {}'.format(self._to_datetime(task.succeeded)))25        print('Failed: {}'.format(self._to_datetime(task.failed)))26        print('Retried: {}'.format(self._to_datetime(task.retried)))27        print('Revoked: {}'.format(self._to_datetime(task.revoked)))28        print('args (arguments): {}'.format(task.args))29        print('kwargs (keyword arguments): {}'.format(task.kwargs))30        print('ETA (Estimated Time of Arrival): {}'.format(task.eta))31        print('Expires: {}'.format(task.expires))32        print('Retries: {}'.format(task.retries))33        print('Worker: {}'.format(task.worker))34        print('Result: {}'.format(task.result))35        print('Exception: {}'.format(task.exception))36        print('Timestamp: {}'.format(self._to_datetime(task.timestamp)))37        print('Runtime: {}'.format(task.runtime))38        print('Traceback: {}'.format(task.traceback))39        print('Exchange: {}'.format(task.exchange))40        print('Routing Key: {}'.format(task.routing_key))41        print('Clock: {}'.format(task.clock))42        print('Client: {}'.format(task.client))43        print('Root: {}'.format(task.root))44        print('Root ID: {}'.format(task.root_id))45        print('Parent: {}'.format(task.parent))46        print('Parent ID: {}'.format(task.parent_id))47        print('Children:')48        for child in task.children:49            print('t{}n'.format(str(child)))50    def _to_datetime(self, timestamp):51        return dt.fromtimestamp(timestamp) if timestamp is not None else None 52class CeleryEventsHandler:53    def __init__(self, celery_app, verbose_logging=False):54        self._app = celery_app55        self._state = celery_app.events.State()56        self._logger = Logger()57        self._verbose_logging = verbose_logging58    def _event_handler(handler):59        def wrapper(self, event):60            self._state.event(event)61            task = self._state.tasks.get(event['uuid'])62            self._logger.log_task_status_change(task, event)63            if(self._verbose_logging):64                self._logger.log_event_details(event)...

datetime.py

Source:datetime.py

...9    timedelta as _timedelta,10)11from .. import compat12ZERO_DELTA = _timedelta()13def _to_datetime(obj):14    if isinstance(obj, datetime):15        return obj16    if isinstance(obj, date):17        return datetime.combine(obj, time(0))18    if isinstance(obj, compat.inttype):19        return datetime(obj, 1, 1)20    if isinstance(obj, (list, tuple)):21        return datetime(*obj)22    if isinstance(obj, dict):23        return datetime(**obj)24    if isinstance(obj, compat.basestring):25        obj = re.findall(r"[0-9]+", obj)26        return datetime(*map(int, obj))27def choice(start, end):28    """random choice between start and end.29    :param datetime start: datetime from.30    :param datetime end: datetime to.31    :return: datetime object that randomly selected32    """33    start = _to_datetime(start)34    end = _to_datetime(end)35    seconds_f = (end - start).total_seconds()36    seconds_i = int(seconds_f)37    chosen = random.randint(0, seconds_i)38    if seconds_i == chosen:39        microseconds = random.randint(0, int((seconds_f - seconds_i) * 1000000))40    else:41        microseconds = random.randint(0, 999999)42    return start + compat.timedelta(seconds=chosen, microseconds=microseconds)43def generator(start, end=None, interval=compat.timedelta(days=1), **noise):44    """datetime generator45    :param datetime start: datetime from46    :param timedelta interval: increase interval47    :param **noise: timedelta args(days, hours, minutes, seconds, microseconds)48    :return: iterator object49    """50    dt = _to_datetime(start)51    end = _to_datetime(end)52    if end:53        direction = (end - dt) >= ZERO_DELTA54    while end is None or (end - dt >= ZERO_DELTA) is direction or end == dt:55        _noise = {}56        for k, v in noise.items():57            if isinstance(v, tuple):58                _noise[k] = random.randint(*v)59            else:60                _noise[k] = random.randint(-v, v)61        yield dt + compat.timedelta(**_noise)62        dt += interval63def range(start, end, interval=compat.timedelta(days=1), callback=None, **noise):64    """create datetime list65    :param datetime start: datetime from66    :param datetiem end: datetime to67    :param timedelta interval: increase interval68    :param callable callback: callback that each element69    :param **noise: timedelta args(days, hours, minutes, seconds, microseconds)70    :return: list71    """72    assert end is not None73    start = _to_datetime(start)74    end = _to_datetime(end)...

logger.py

Source:logger.py

1from datetime import datetime as dt2class Logger:3    def log_task_status_change(self, task, event):4        print('[{}] {} {} (STATE={}, UUID={})'.format(5            self._to_datetime(task.timestamp),6            event['type'].upper(),7            task.name,8            task.state.upper(),9            task.uuid10        ))11    def log_event_details(self, event):12        print('EVENT DETAILS: {}'.format(event))13    def log_task_details(self, task):14        print('TASK DETAILS:')15        print('UUID: {}'.format(task.uuid))16        print('Name: {}'.format(task.name))17        print('State: {}'.format(task.state))18        print('Received: {}'.format(self._to_datetime(task.received)))19        print('Sent: {}'.format(self._to_datetime(task.sent)))20        print('Started: {}'.format(self._to_datetime(task.started)))21        print('Rejected: {}'.format(self._to_datetime(task.rejected)))22        print('Succeeded: {}'.format(self._to_datetime(task.succeeded)))23        print('Failed: {}'.format(self._to_datetime(task.failed)))24        print('Retried: {}'.format(self._to_datetime(task.retried)))25        print('Revoked: {}'.format(self._to_datetime(task.revoked)))26        print('args (arguments): {}'.format(task.args))27        print('kwargs (keyword arguments): {}'.format(task.kwargs))28        print('ETA (Estimated Time of Arrival): {}'.format(task.eta))29        print('Expires: {}'.format(task.expires))30        print('Retries: {}'.format(task.retries))31        print('Worker: {}'.format(task.worker))32        print('Result: {}'.format(task.result))33        print('Exception: {}'.format(task.exception))34        print('Timestamp: {}'.format(self._to_datetime(task.timestamp)))35        print('Runtime: {}'.format(task.runtime))36        print('Traceback: {}'.format(task.traceback))37        print('Exchange: {}'.format(task.exchange))38        print('Routing Key: {}'.format(task.routing_key))39        print('Clock: {}'.format(task.clock))40        print('Client: {}'.format(task.client))41        print('Root: {}'.format(task.root))42        print('Root ID: {}'.format(task.root_id))43        print('Parent: {}'.format(task.parent))44        print('Parent ID: {}'.format(task.parent_id))45        print('Children:')46        for child in task.children:47            print('t{}n'.format(str(child)))48        49        print("")50    def _to_datetime(self, timestamp):...

Automation Testing Tutorials

Learn to execute automation testing from scratch with LambdaTest Learning Hub. Right from setting up the prerequisites to run your first automation test, to following best practices and diving deeper into advanced test scenarios. LambdaTest Learning Hubs compile a list of step-by-step guides to help you be proficient with different test automation frameworks i.e. Selenium, Cypress, TestNG etc.