Python datetime.datetime 模块,max() 实例源码

我们从Python开源项目中,提取了以下50个代码示例,用于说明如何使用datetime.datetime.max()

项目:ouroboros    作者:pybee    | 项目源码 | 文件源码
def test_overflow(self):
        tiny = timedelta.resolution

        td = timedelta.min + tiny
        td -= tiny  # no problem
        self.assertRaises(OverflowError, td.__sub__, tiny)
        self.assertRaises(OverflowError, td.__add__, -tiny)

        td = timedelta.max - tiny
        td += tiny  # no problem
        self.assertRaises(OverflowError, td.__add__, tiny)
        self.assertRaises(OverflowError, td.__sub__, -tiny)

        self.assertRaises(OverflowError, lambda: -timedelta.max)

        day = timedelta(1)
        self.assertRaises(OverflowError, day.__mul__, 10**9)
        self.assertRaises(OverflowError, day.__mul__, 1e9)
        self.assertRaises(OverflowError, day.__truediv__, 1e-20)
        self.assertRaises(OverflowError, day.__truediv__, 1e-10)
        self.assertRaises(OverflowError, day.__truediv__, 9e-10)
项目:shifthelper    作者:fact-project    | 项目源码 | 文件源码
def update_heartbeat():
    @retry(
        stop_max_delay=30000,  # 30 seconds max
        wait_exponential_multiplier=100,  # wait 2^i * 100 ms, on the i-th retry
        wait_exponential_max=1000,  # but wait 1 second per try maximum
        wrap_exception=True
    )
    def retry_fetch_fail_after_30sec():
        return requests.post(
            config['webservice']['shifthelperHeartbeat'],
            auth=(
                config['webservice']['user'],
                config['webservice']['password']
            )
        ).json()
    try:
        return retry_fetch_fail_after_30sec()
    except RetryError as e:
        return {}
项目:facebook-message-analysis    作者:szheng17    | 项目源码 | 文件源码
def filter_by_first_message_dt(self, start_dt=datetime.min, end_dt=datetime.max):
        """
        Returns a copy of self after filtering self.conversations by
        conversations whose first messages lie in a datetime interval.

        Args:
            start_dt: A datetime object satisfying start_dt <= end_dt.
            end_dt: A datetime object satisfying start_dt <= end_dt.

        Returns:
            A UserConversations object that is equal to self after filtering
                self.conversations such that the datetime of the first message
                in each conversation is in the closed interval
                [start_dt, end_dt].

        Raises:
            EmptyUserConversationsError: Filtering self.conversations results
                in an empty list.
        """
        if start_dt > end_dt:
            raise ValueError('Must have start_dt <= end_dt')
        conversation_filter = lambda x: x.messages[0].timestamp >= start_dt and x.messages[0].timestamp <= end_dt
        return self.filter_user_conversations(conversation_filter)
项目:facebook-message-analysis    作者:szheng17    | 项目源码 | 文件源码
def filter_by_last_message_dt(self, start_dt=datetime.min, end_dt=datetime.max):
        """
        Returns a copy of self after filtering self.conversations by
        conversations whose last messages lie in a datetime interval.

        Args:
            start_dt: A datetime object satisfying start_dt <= end_dt.
            end_dt: A datetime object satisfying start_dt <= end_dt.

        Returns:
            A UserConversations object that is equal to self after filtering
                self.conversations such that the datetime of the last message
                in each conversation is in the closed interval
                [start_dt, end_dt].

        Raises:
            EmptyUserConversationsError: Filtering self.conversations results
                in an empty list.
        """
        if start_dt > end_dt:
            raise ValueError('Must have start_dt <= end_dt')
        conversation_filter = lambda x: x.messages[-1].timestamp >= start_dt and x.messages[-1].timestamp <= end_dt
        return self.filter_user_conversations(conversation_filter)
项目:facebook-message-analysis    作者:szheng17    | 项目源码 | 文件源码
def filter_by_datetime(self, start_dt=datetime.min, end_dt=datetime.max):
        """
        Returns a copy of self after filtering each conversation's messages
        to only contain messages that lie in a datetime interval and removing
        conversations with no messages in the datetime interval.

        Args:
            start_dt: A datetime object satisfying start_dt <= end_dt.
            end_dt: A datetime object satisfying start_dt <= end_dt.

        Returns:
            A UserConversations object that is equal to self after filtering
                the messages of each conversation in self.conversations to keep
                messages whose datetimes are in the closed interval
                [start_dt, end_dt]. Conversations with no messages in the
                interval are discarded.

        Raises:
            EmptyUserConversationsError: Across all conversations, there is no
                message that lies in the closed interval [start_dt, end_dt].
        """
        if start_dt > end_dt:
            raise ValueError("start_dt must be less than or equal to end_dt")
        message_filter = lambda x: x.timestamp >= start_dt and x.timestamp <= end_dt
        return self.filter_conversations(message_filter=message_filter)
项目:facebook-message-analysis    作者:szheng17    | 项目源码 | 文件源码
def filter_by_datetime(self, start_dt=datetime.min, end_dt=datetime.max):
        """
        Returns a copy of self after filtering self.messages by messages that
        lie in a datetime interval.

        Args:
            start_dt: A datetime object satisfying start_dt <= end_dt.
            end_dt: A datetime object satisfying start_dt <= end_dt.

        Returns:
            A Conversation object that is equal to self after filtering by
                messages whose datetimes are in the closed interval
                [start_dt, end_dt].

        Raises:
            EmptyConversationError: Filtering self.messages results in an empty
                list.
        """
        if start_dt > end_dt:
            raise ValueError("start_dt must be less than or equal to end_dt")
        message_filter = lambda x: x.timestamp >= start_dt and x.timestamp <= end_dt
        filtered = self.filter_conversation(message_filter=message_filter)
        return filtered
项目:slack-sql    作者:wang502    | 项目源码 | 文件源码
def cast_date(value, connection):
    """Cast a date value."""
    # The output format depends on the server setting DateStyle.  The default
    # setting ISO and the setting for German are actually unambiguous.  The
    # order of days and months in the other two settings is however ambiguous,
    # so at least here we need to consult the setting to properly parse values.
    if value == '-infinity':
        return date.min
    if value == 'infinity':
        return date.max
    value = value.split()
    if value[-1] == 'BC':
        return date.min
    value = value[0]
    if len(value) > 10:
        return date.max
    fmt = connection.date_format()
    return datetime.strptime(value, fmt).date()
项目:slack-sql    作者:wang502    | 项目源码 | 文件源码
def cast_timestamp(value, connection):
    """Cast a timestamp value."""
    if value == '-infinity':
        return datetime.min
    if value == 'infinity':
        return datetime.max
    value = value.split()
    if value[-1] == 'BC':
        return datetime.min
    fmt = connection.date_format()
    if fmt.endswith('-%Y') and len(value) > 2:
        value = value[1:5]
        if len(value[3]) > 4:
            return datetime.max
        fmt = ['%d %b' if fmt.startswith('%d') else '%b %d',
            '%H:%M:%S.%f' if len(value[2]) > 8 else '%H:%M:%S', '%Y']
    else:
        if len(value[0]) > 10:
            return datetime.max
        fmt = [fmt, '%H:%M:%S.%f' if len(value[1]) > 8 else '%H:%M:%S']
    return datetime.strptime(' '.join(value), ' '.join(fmt))
项目:slack-sql    作者:wang502    | 项目源码 | 文件源码
def cast_timestamptz(value, connection):
    """Cast a timestamptz value."""
    if value == '-infinity':
        return datetime.min
    if value == 'infinity':
        return datetime.max
    value = value.split()
    if value[-1] == 'BC':
        return datetime.min
    fmt = connection.date_format()
    if fmt.endswith('-%Y') and len(value) > 2:
        value = value[1:]
        if len(value[3]) > 4:
            return datetime.max
        fmt = ['%d %b' if fmt.startswith('%d') else '%b %d',
            '%H:%M:%S.%f' if len(value[2]) > 8 else '%H:%M:%S', '%Y']
        value, tz = value[:-1], value[-1]
    else:
        if fmt.startswith('%Y-'):
            tz = _re_timezone.match(value[1])
            if tz:
                value[1], tz = tz.groups()
            else:
                tz = '+0000'
        else:
            value, tz = value[:-1], value[-1]
        if len(value[0]) > 10:
            return datetime.max
        fmt = [fmt, '%H:%M:%S.%f' if len(value[1]) > 8 else '%H:%M:%S']
    if _has_timezone:
        value.append(_timezone_as_offset(tz))
        fmt.append('%z')
        return datetime.strptime(' '.join(value), ' '.join(fmt))
    return datetime.strptime(' '.join(value), ' '.join(fmt)).replace(
        tzinfo=_get_timezone(tz))
项目:slack-sql    作者:wang502    | 项目源码 | 文件源码
def cast_date(value, connection):
    """Cast a date value."""
    # The output format depends on the server setting DateStyle.  The default
    # setting ISO and the setting for German are actually unambiguous.  The
    # order of days and months in the other two settings is however ambiguous,
    # so at least here we need to consult the setting to properly parse values.
    if value == '-infinity':
        return date.min
    if value == 'infinity':
        return date.max
    value = value.split()
    if value[-1] == 'BC':
        return date.min
    value = value[0]
    if len(value) > 10:
        return date.max
    fmt = connection.date_format()
    return datetime.strptime(value, fmt).date()
项目:slack-sql    作者:wang502    | 项目源码 | 文件源码
def cast_timestamp(value, connection):
    """Cast a timestamp value."""
    if value == '-infinity':
        return datetime.min
    if value == 'infinity':
        return datetime.max
    value = value.split()
    if value[-1] == 'BC':
        return datetime.min
    fmt = connection.date_format()
    if fmt.endswith('-%Y') and len(value) > 2:
        value = value[1:5]
        if len(value[3]) > 4:
            return datetime.max
        fmt = ['%d %b' if fmt.startswith('%d') else '%b %d',
            '%H:%M:%S.%f' if len(value[2]) > 8 else '%H:%M:%S', '%Y']
    else:
        if len(value[0]) > 10:
            return datetime.max
        fmt = [fmt, '%H:%M:%S.%f' if len(value[1]) > 8 else '%H:%M:%S']
    return datetime.strptime(' '.join(value), ' '.join(fmt))
项目:slack-sql    作者:wang502    | 项目源码 | 文件源码
def testTruncateRestart(self):
        truncate = self.db.truncate
        self.assertRaises(TypeError, truncate, 'test_table', restart='invalid')
        query = self.db.query
        self.createTable('test_table', 'n serial, t text')
        for n in range(3):
            query("insert into test_table (t) values ('test')")
        q = "select count(n), min(n), max(n) from test_table"
        r = query(q).getresult()[0]
        self.assertEqual(r, (3, 1, 3))
        truncate('test_table')
        r = query(q).getresult()[0]
        self.assertEqual(r, (0, None, None))
        for n in range(3):
            query("insert into test_table (t) values ('test')")
        r = query(q).getresult()[0]
        self.assertEqual(r, (3, 4, 6))
        truncate('test_table', restart=True)
        r = query(q).getresult()[0]
        self.assertEqual(r, (0, None, None))
        for n in range(3):
            query("insert into test_table (t) values ('test')")
        r = query(q).getresult()[0]
        self.assertEqual(r, (3, 1, 3))
项目:slack-sql    作者:wang502    | 项目源码 | 文件源码
def testDate(self):
        query = self.db.query
        for datestyle in ('ISO', 'Postgres, MDY', 'Postgres, DMY',
                'SQL, MDY', 'SQL, DMY', 'German'):
            self.db.set_parameter('datestyle', datestyle)
            d = date(2016, 3, 14)
            q = "select $1::date"
            r = query(q, (d,)).getresult()[0][0]
            self.assertIsInstance(r, date)
            self.assertEqual(r, d)
            q = "select '10000-08-01'::date, '0099-01-08 BC'::date"
            r = query(q).getresult()[0]
            self.assertIsInstance(r[0], date)
            self.assertIsInstance(r[1], date)
            self.assertEqual(r[0], date.max)
            self.assertEqual(r[1], date.min)
        q = "select 'infinity'::date, '-infinity'::date"
        r = query(q).getresult()[0]
        self.assertIsInstance(r[0], date)
        self.assertIsInstance(r[1], date)
        self.assertEqual(r[0], date.max)
        self.assertEqual(r[1], date.min)
项目:slack-sql    作者:wang502    | 项目源码 | 文件源码
def cast_date(value, connection):
    """Cast a date value."""
    # The output format depends on the server setting DateStyle.  The default
    # setting ISO and the setting for German are actually unambiguous.  The
    # order of days and months in the other two settings is however ambiguous,
    # so at least here we need to consult the setting to properly parse values.
    if value == '-infinity':
        return date.min
    if value == 'infinity':
        return date.max
    value = value.split()
    if value[-1] == 'BC':
        return date.min
    value = value[0]
    if len(value) > 10:
        return date.max
    fmt = connection.date_format()
    return datetime.strptime(value, fmt).date()
项目:slack-sql    作者:wang502    | 项目源码 | 文件源码
def cast_date(value, connection):
    """Cast a date value."""
    # The output format depends on the server setting DateStyle.  The default
    # setting ISO and the setting for German are actually unambiguous.  The
    # order of days and months in the other two settings is however ambiguous,
    # so at least here we need to consult the setting to properly parse values.
    if value == '-infinity':
        return date.min
    if value == 'infinity':
        return date.max
    value = value.split()
    if value[-1] == 'BC':
        return date.min
    value = value[0]
    if len(value) > 10:
        return date.max
    fmt = connection.date_format()
    return datetime.strptime(value, fmt).date()
项目:slack-sql    作者:wang502    | 项目源码 | 文件源码
def cast_timestamp(value, connection):
    """Cast a timestamp value."""
    if value == '-infinity':
        return datetime.min
    if value == 'infinity':
        return datetime.max
    value = value.split()
    if value[-1] == 'BC':
        return datetime.min
    fmt = connection.date_format()
    if fmt.endswith('-%Y') and len(value) > 2:
        value = value[1:5]
        if len(value[3]) > 4:
            return datetime.max
        fmt = ['%d %b' if fmt.startswith('%d') else '%b %d',
            '%H:%M:%S.%f' if len(value[2]) > 8 else '%H:%M:%S', '%Y']
    else:
        if len(value[0]) > 10:
            return datetime.max
        fmt = [fmt, '%H:%M:%S.%f' if len(value[1]) > 8 else '%H:%M:%S']
    return datetime.strptime(' '.join(value), ' '.join(fmt))
项目:zippy    作者:securesystemslab    | 项目源码 | 文件源码
def test_overflow(self):
        tiny = timedelta.resolution

        td = timedelta.min + tiny
        td -= tiny  # no problem
        self.assertRaises(OverflowError, td.__sub__, tiny)
        self.assertRaises(OverflowError, td.__add__, -tiny)

        td = timedelta.max - tiny
        td += tiny  # no problem
        self.assertRaises(OverflowError, td.__add__, tiny)
        self.assertRaises(OverflowError, td.__sub__, -tiny)

        self.assertRaises(OverflowError, lambda: -timedelta.max)

        day = timedelta(1)
        self.assertRaises(OverflowError, day.__mul__, 10**9)
        self.assertRaises(OverflowError, day.__mul__, 1e9)
        self.assertRaises(OverflowError, day.__truediv__, 1e-20)
        self.assertRaises(OverflowError, day.__truediv__, 1e-10)
        self.assertRaises(OverflowError, day.__truediv__, 9e-10)
项目:auxi.0    作者:Ex-Mente    | 项目源码 | 文件源码
def transaction_list(self, start=datetime.min,
                         end=datetime.max,
                         format=ReportFormat.printout,
                         component_path="",
                         output_path=None):
        """
        Generate a transaction list report.

        :param start: The start date to generate the report for.
        :param end: The end date to generate the report for.
        :param format: The format of the report.
        :param component_path: The path of the component to filter the report's
          transactions by.
        :param output_path: The path to the file the report is written to.
          If None, then the report is not written to a file.

        :returns: The generated report.
        """

        rpt = TransactionList(self, start, end, component_path, output_path)
        return rpt.render(format)
项目:auxi.0    作者:Ex-Mente    | 项目源码 | 文件源码
def income_statement(self, start=datetime.min,
                         end=datetime.max,
                         format=ReportFormat.printout,
                         component_path="",
                         output_path=None):
        """
        Generate a transaction list report.

        :param start: The start date to generate the report for.
        :param end: The end date to generate the report for.
        :param format: The format of the report.
        :param component_path: The path of the component to filter the report's
          transactions by.
        :param output_path: The path to the file the report is written to.
          If None, then the report is not written to a file.

        :returns: The generated report.
        """

        rpt = IncomeStatement(self, start, end, component_path, output_path)
        return rpt.render(format)
项目:auxi.0    作者:Ex-Mente    | 项目源码 | 文件源码
def __init__(self, name,
                 dt_account,
                 cr_account,
                 amount=0,
                 start=datetime.min,
                 end=datetime.max,
                 interval=1,
                 description=None):
        """
        """
        super(BasicActivity, self).__init__(
            name,
            description=description,
            start=start,
            end=end,
            interval=interval)
        self.interval = interval
        self.amount = amount
        self.dt_account = dt_account
        self.cr_account = cr_account
项目:openregistry.api    作者:openprocurement    | 项目源码 | 文件源码
def test_IsoDateTimeType_model(self):

        dt = IsoDateTimeType()

        value = dt.to_primitive(now)
        self.assertEqual(now, dt.to_native(now))
        self.assertEqual(now, dt.to_native(value))

        date = datetime.now()
        value = dt.to_primitive(date)
        self.assertEqual(date, dt.to_native(date))
        self.assertEqual(now.tzinfo, dt.to_native(value).tzinfo)

        # ParseError
        for date in (None, '', 2017, "2007-06-23X06:40:34.00Z"):
            with self.assertRaisesRegexp(ConversionError, 
                      u'Could not parse %s. Should be ISO8601.' % date):
                dt.to_native(date)
        # OverflowError
        for date in (datetime.max, datetime.min):
            self.assertEqual(date, dt.to_native(date))
            with self.assertRaises(ConversionError):
                dt.to_native(dt.to_primitive(date))
项目:web_ctp    作者:molebot    | 项目源码 | 文件源码
def test_overflow(self):
        tiny = timedelta.resolution

        td = timedelta.min + tiny
        td -= tiny  # no problem
        self.assertRaises(OverflowError, td.__sub__, tiny)
        self.assertRaises(OverflowError, td.__add__, -tiny)

        td = timedelta.max - tiny
        td += tiny  # no problem
        self.assertRaises(OverflowError, td.__add__, tiny)
        self.assertRaises(OverflowError, td.__sub__, -tiny)

        self.assertRaises(OverflowError, lambda: -timedelta.max)

        day = timedelta(1)
        self.assertRaises(OverflowError, day.__mul__, 10**9)
        self.assertRaises(OverflowError, day.__mul__, 1e9)
        self.assertRaises(OverflowError, day.__truediv__, 1e-20)
        self.assertRaises(OverflowError, day.__truediv__, 1e-10)
        self.assertRaises(OverflowError, day.__truediv__, 9e-10)
项目:sdk    作者:vt-dev    | 项目源码 | 文件源码
def map_from_json(json_slots):
        slots = []
        for json_slot in json_slots:
            from_long = json_slot['start'] / 1e3
            if from_long < TimeSlot.time_to_long(datetime.min):
                from_long = TimeSlot.time_to_long(datetime.min)

            duration_long = json_slot['duration'] / 1e3
            to_long = from_long + duration_long
            if to_long > TimeSlot.time_to_long(datetime.max):
                to_long = TimeSlot.time_to_long(datetime.max)

            fr = datetime.fromtimestamp(from_long)
            to = datetime.fromtimestamp(to_long)
            slots.append(TimeSlot(fr, to))
        return slots
项目:python-cache    作者:python-cache    | 项目源码 | 文件源码
def set(self, key, value):
        """Sets the key and value represented by this cache item.

        Note:
            The $value argument may be any item that can be serialized by python,
            although the method of serialization is left up to the Implementing
            Library.

        Args:
            value: The serializable value to be stored.

        :return The invoked object.

        """
        self.key = key
        self.value = value
        self.expire_at = datetime.max
        self._setHit()
        return self.key, self.value
项目:python-cache    作者:python-cache    | 项目源码 | 文件源码
def save(self, item):
        """Persists a cache item immediately.

        :param item: The cache item to save.

        :return True if the item was successfully persisted. False if there was an error.

        """

        if item.expire_at != datetime.max:
            expire_seconds = (item.expire_at - datetime.utcnow()).seconds
            if expire_seconds > 0:
                return self.client.setex(self.normalize_key(item.key),
                                     cPickle.dumps(item), expire_seconds)
            else:
                return False
        else:
            return self.client.set(self.normalize_key(item.key),
                                   cPickle.dumps(item))
项目:nav    作者:UNINETT    | 项目源码 | 文件源码
def _load_existing_mappings(self):
        """Load the existing ARP records for this box from the db.

        Returns:

          A deferred whose result is a dictionary: { (ip, mac): arpid }
        """
        self._logger.debug("Loading open arp records from database")
        open_arp_records_queryset = manage.Arp.objects.filter(
            netbox__id=self.netbox.id,
            end_time__gte=datetime.max).values('id', 'ip', 'mac')
        open_arp_records = yield db.run_in_thread(
            storage.shadowify_queryset_and_commit,
            open_arp_records_queryset)
        self._logger.debug("Loaded %d open records from arp",
                           len(open_arp_records))

        open_mappings = dict(((IP(arp['ip']), arp['mac']), arp['id'])
                             for arp in open_arp_records)
        defer.returnValue(open_mappings)
项目:nav    作者:UNINETT    | 项目源码 | 文件源码
def _make_new_mappings(self, mappings):
        """Convert a sequence of (ip, mac) tuples into a Arp shadow containers.

        Arguments:

          mappings -- An iterable containing tuples: (ip, mac)

        """
        netbox = self.containers.factory(None, shadows.Netbox)
        timestamp = datetime.now()
        infinity = datetime.max

        for (ip, mac) in mappings:
            if not ip or not mac:
                continue  # Some devices seem to return empty results!
            arp = self.containers.factory((ip, mac), shadows.Arp)
            arp.netbox = netbox
            arp.sysname = self.netbox.sysname
            arp.ip = ip.strCompressed()
            arp.mac = mac
            arp.prefix_id = self._find_largest_matching_prefix(ip)
            arp.start_time = timestamp
            arp.end_time = infinity
项目:nav    作者:UNINETT    | 项目源码 | 文件源码
def test_time_since(self):
        def timestamp_calc(*args, **kwargs):
            return datetime.now() - timedelta(*args, **kwargs)

        minute = 60
        hour = minute * 60

        self.assertEqual(
            time_since(None),
            "Never")
        self.assertEqual(
            time_since(timestamp_calc(seconds=(10 * minute + 10))),
            u"10\xa0mins")
        self.assertEqual(
            time_since(timestamp_calc(seconds=(1 * minute + 5))),
            u"1\xa0min")
        self.assertEqual(
            time_since(timestamp_calc(0)),
            u"Now")
        self.assertEqual(
            time_since(datetime.max),
            u"Now")
项目:kbe_server    作者:xiaohaoppy    | 项目源码 | 文件源码
def test_overflow(self):
        tiny = timedelta.resolution

        td = timedelta.min + tiny
        td -= tiny  # no problem
        self.assertRaises(OverflowError, td.__sub__, tiny)
        self.assertRaises(OverflowError, td.__add__, -tiny)

        td = timedelta.max - tiny
        td += tiny  # no problem
        self.assertRaises(OverflowError, td.__add__, tiny)
        self.assertRaises(OverflowError, td.__sub__, -tiny)

        self.assertRaises(OverflowError, lambda: -timedelta.max)

        day = timedelta(1)
        self.assertRaises(OverflowError, day.__mul__, 10**9)
        self.assertRaises(OverflowError, day.__mul__, 1e9)
        self.assertRaises(OverflowError, day.__truediv__, 1e-20)
        self.assertRaises(OverflowError, day.__truediv__, 1e-10)
        self.assertRaises(OverflowError, day.__truediv__, 9e-10)
项目:idealoom    作者:conversence    | 项目源码 | 文件源码
def has_notification(self):
        settings = self.settings_json
        notifications = settings.get('notifications', [])
        now = datetime.utcnow()

        for notification in notifications:
            try:
                start = parse_datetime(notification['start'])
                end = notification.get('end', None)
                end = parse_datetime(end) if end else datetime.max
                if now < start or now > end:
                    continue
            except (ValueError, TypeError, KeyError) as e:
                continue
            notification_data = self.notification_data(notification)
            if notification_data:
                yield notification_data
项目:pygresql    作者:Cito    | 项目源码 | 文件源码
def cast_date(value, connection):
    """Cast a date value."""
    # The output format depends on the server setting DateStyle.  The default
    # setting ISO and the setting for German are actually unambiguous.  The
    # order of days and months in the other two settings is however ambiguous,
    # so at least here we need to consult the setting to properly parse values.
    if value == '-infinity':
        return date.min
    if value == 'infinity':
        return date.max
    value = value.split()
    if value[-1] == 'BC':
        return date.min
    value = value[0]
    if len(value) > 10:
        return date.max
    fmt = connection.date_format()
    return datetime.strptime(value, fmt).date()
项目:pygresql    作者:Cito    | 项目源码 | 文件源码
def cast_timestamp(value, connection):
    """Cast a timestamp value."""
    if value == '-infinity':
        return datetime.min
    if value == 'infinity':
        return datetime.max
    value = value.split()
    if value[-1] == 'BC':
        return datetime.min
    fmt = connection.date_format()
    if fmt.endswith('-%Y') and len(value) > 2:
        value = value[1:5]
        if len(value[3]) > 4:
            return datetime.max
        fmt = ['%d %b' if fmt.startswith('%d') else '%b %d',
            '%H:%M:%S.%f' if len(value[2]) > 8 else '%H:%M:%S', '%Y']
    else:
        if len(value[0]) > 10:
            return datetime.max
        fmt = [fmt, '%H:%M:%S.%f' if len(value[1]) > 8 else '%H:%M:%S']
    return datetime.strptime(' '.join(value), ' '.join(fmt))
项目:pygresql    作者:Cito    | 项目源码 | 文件源码
def cast_date(value, connection):
    """Cast a date value."""
    # The output format depends on the server setting DateStyle.  The default
    # setting ISO and the setting for German are actually unambiguous.  The
    # order of days and months in the other two settings is however ambiguous,
    # so at least here we need to consult the setting to properly parse values.
    if value == '-infinity':
        return date.min
    if value == 'infinity':
        return date.max
    value = value.split()
    if value[-1] == 'BC':
        return date.min
    value = value[0]
    if len(value) > 10:
        return date.max
    fmt = connection.date_format()
    return datetime.strptime(value, fmt).date()
项目:pygresql    作者:Cito    | 项目源码 | 文件源码
def cast_timestamp(value, connection):
    """Cast a timestamp value."""
    if value == '-infinity':
        return datetime.min
    if value == 'infinity':
        return datetime.max
    value = value.split()
    if value[-1] == 'BC':
        return datetime.min
    fmt = connection.date_format()
    if fmt.endswith('-%Y') and len(value) > 2:
        value = value[1:5]
        if len(value[3]) > 4:
            return datetime.max
        fmt = ['%d %b' if fmt.startswith('%d') else '%b %d',
            '%H:%M:%S.%f' if len(value[2]) > 8 else '%H:%M:%S', '%Y']
    else:
        if len(value[0]) > 10:
            return datetime.max
        fmt = [fmt, '%H:%M:%S.%f' if len(value[1]) > 8 else '%H:%M:%S']
    return datetime.strptime(' '.join(value), ' '.join(fmt))
项目:pygresql    作者:Cito    | 项目源码 | 文件源码
def testTruncateRestart(self):
        truncate = self.db.truncate
        self.assertRaises(TypeError, truncate, 'test_table', restart='invalid')
        query = self.db.query
        self.createTable('test_table', 'n serial, t text')
        for n in range(3):
            query("insert into test_table (t) values ('test')")
        q = "select count(n), min(n), max(n) from test_table"
        r = query(q).getresult()[0]
        self.assertEqual(r, (3, 1, 3))
        truncate('test_table')
        r = query(q).getresult()[0]
        self.assertEqual(r, (0, None, None))
        for n in range(3):
            query("insert into test_table (t) values ('test')")
        r = query(q).getresult()[0]
        self.assertEqual(r, (3, 4, 6))
        truncate('test_table', restart=True)
        r = query(q).getresult()[0]
        self.assertEqual(r, (0, None, None))
        for n in range(3):
            query("insert into test_table (t) values ('test')")
        r = query(q).getresult()[0]
        self.assertEqual(r, (3, 1, 3))
项目:pygresql    作者:Cito    | 项目源码 | 文件源码
def testDate(self):
        query = self.db.query
        for datestyle in ('ISO', 'Postgres, MDY', 'Postgres, DMY',
                'SQL, MDY', 'SQL, DMY', 'German'):
            self.db.set_parameter('datestyle', datestyle)
            d = date(2016, 3, 14)
            q = "select $1::date"
            r = query(q, (d,)).getresult()[0][0]
            self.assertIsInstance(r, date)
            self.assertEqual(r, d)
            q = "select '10000-08-01'::date, '0099-01-08 BC'::date"
            r = query(q).getresult()[0]
            self.assertIsInstance(r[0], date)
            self.assertIsInstance(r[1], date)
            self.assertEqual(r[0], date.max)
            self.assertEqual(r[1], date.min)
        q = "select 'infinity'::date, '-infinity'::date"
        r = query(q).getresult()[0]
        self.assertIsInstance(r[0], date)
        self.assertIsInstance(r[1], date)
        self.assertEqual(r[0], date.max)
        self.assertEqual(r[1], date.min)
项目:facebook-message-analysis    作者:szheng17    | 项目源码 | 文件源码
def get_user_to_damped_n_messages(self, dt_max, alpha):
        """
        Maps each user to the number of messages before a reference datetime, 
        where each message count is exponentially damped by a constant times
        the difference between the reference datetime and the datetime of the
        message.

        Args:
            dt_max: A datetime representing the max datetime of messages
                to consider.
            alpha: A nonnegative float representing the damping factor.

        Returns:
            user_to_damped_n_messages: A dict mapping each user in
                self.users_union to the damped number of messages by that user
                before dt_max. The contribution of a message is a float equal
                to exp(-alpha * t), where t is the difference in days between
                dt_max and the datetime of the message.
        """
        if alpha < 0:
            raise ValueError('Must have alpha >= 0')
        try:
            # Only keep messages with datetimes <= dt_max
            filtered = self.filter_by_datetime(end_dt=dt_max)
        except EmptyConversationError:
            # Map all users to 0 if dt_max occurs before all messages
            return self.get_user_to_message_statistic(lambda x: 0)
        damped_message_count = lambda x: self.exp_damped_day_difference(dt_max, x.timestamp, alpha)
        user_to_damped_n_messages = filtered.get_user_to_message_statistic(damped_message_count)
        return user_to_damped_n_messages
项目:black-hole    作者:liuhao1024    | 项目源码 | 文件源码
def tslice(unit, start=ben(), end=None, step=1, count=float('inf')):
    """
    tslice(unit, start=None, end=None, step=1, count=None) -> generator of Blackhole object
    unit in ['year', 'month', 'day', 'hour', 'minute', 'second', 'microsecond']
    this is some kind xrange-like
    :param unit:
    :param start:
    :param end:
    :param step:
    :param count:
    :return:
    """
    if unit not in Blackhole._units:
        raise AttributeError()
    if isinstance(start, basestring):
        start = ben(start)
    if isinstance(end, basestring):
        end = ben(end)

    cnt = 0
    if step > 0:
        end = end or ben(datetime.max)
        while start < end and cnt < count:
            yield start
            start = start.shifted(**{unit: step})
            cnt += 1
    elif step < 0:
        end = end or ben(datetime.min)
        while start > end and cnt < count:
            yield start
            start = start.shifted(**{unit: step})
            cnt += 1
项目:python-zhmcclient    作者:zhmcclient    | 项目源码 | 文件源码
def find_max_value(test_func, initial_value):
    """
    Starting from an initial number (integer or float), find the maximum value
    for which the test function does not yet fail, and return that maximum
    value.
    """
    assert isinstance(initial_value, int) and initial_value > 0

    fails = FailsArray(test_func)
    value = initial_value

    # Advance the value exponentially beyond the max value
    while fails[value] == 0:
        value *= 2

    # Search for the exact max value in the previous range. We search for the
    # boundary where the fails array goes from 0 to 1.
    boundary = 0.5
    value = binary_search(fails, boundary, value // 2, value)
    max_value = value - 1

    # Verify that we found exactly the maximum:
    assert fails[max_value] == 0 and fails[max_value + 1] == 1, \
        "max_value={}, fails[+-2]: {}, {}, {}, {}, {}".\
        format(max_value, fails[max_value - 2], fails[max_value - 1],
               fails[max_value], fails[max_value + 1], fails[max_value + 2])

    return max_value
项目:python-zhmcclient    作者:zhmcclient    | 项目源码 | 文件源码
def x_test_print_datetime_max(self):
        """Print datetime.max."""
        print("\nMax value for Python datetime (datetime.max): {!r}".
              format(datetime.max))
        sys.stdout.flush()
项目:python-zhmcclient    作者:zhmcclient    | 项目源码 | 文件源码
def test_datetime_max(self):
        """Test timestamp_from_datetime() with datetime.max."""

        # The test is that it does not raise an exception:
        timestamp_from_datetime(datetime.max)
项目:Orator-Google-App-Engine    作者:MakarenaLabs    | 项目源码 | 文件源码
def _get_iteration_params(cls, end, limit):

        if end is None:

            if limit is None:
                raise Exception('one of \'end\' or \'limit\' is required')

            return cls.max, limit

        else:
            return end, sys.maxsize
项目:slack-sql    作者:wang502    | 项目源码 | 文件源码
def testTimestamp(self):
        query = self.db.query
        for datestyle in ('ISO', 'Postgres, MDY', 'Postgres, DMY',
                'SQL, MDY', 'SQL, DMY', 'German'):
            self.db.set_parameter('datestyle', datestyle)
            d = datetime(2016, 3, 14)
            q = "select $1::timestamp"
            r = query(q, (d,)).getresult()[0][0]
            self.assertIsInstance(r, datetime)
            self.assertEqual(r, d)
            d = datetime(2016, 3, 14, 15, 9, 26)
            q = "select $1::timestamp"
            r = query(q, (d,)).getresult()[0][0]
            self.assertIsInstance(r, datetime)
            self.assertEqual(r, d)
            d = datetime(2016, 3, 14, 15, 9, 26, 535897)
            q = "select $1::timestamp"
            r = query(q, (d,)).getresult()[0][0]
            self.assertIsInstance(r, datetime)
            self.assertEqual(r, d)
            q = ("select '10000-08-01 AD'::timestamp,"
                " '0099-01-08 BC'::timestamp")
            r = query(q).getresult()[0]
            self.assertIsInstance(r[0], datetime)
            self.assertIsInstance(r[1], datetime)
            self.assertEqual(r[0], datetime.max)
            self.assertEqual(r[1], datetime.min)
        q = "select 'infinity'::timestamp, '-infinity'::timestamp"
        r = query(q).getresult()[0]
        self.assertIsInstance(r[0], datetime)
        self.assertIsInstance(r[1], datetime)
        self.assertEqual(r[0], datetime.max)
        self.assertEqual(r[1], datetime.min)
项目:slack-sql    作者:wang502    | 项目源码 | 文件源码
def testTimestamptz(self):
        query = self.db.query
        timezones = dict(CET=1, EET=2, EST=-5, UTC=0)
        for timezone in sorted(timezones):
            tz = '%+03d00' % timezones[timezone]
            try:
                tzinfo = datetime.strptime(tz, '%z').tzinfo
            except ValueError:  # Python < 3.2
                tzinfo = pg._get_timezone(tz)
            self.db.set_parameter('timezone', timezone)
            for datestyle in ('ISO', 'Postgres, MDY', 'Postgres, DMY',
                    'SQL, MDY', 'SQL, DMY', 'German'):
                self.db.set_parameter('datestyle', datestyle)
                d = datetime(2016, 3, 14, tzinfo=tzinfo)
                q = "select $1::timestamptz"
                r = query(q, (d,)).getresult()[0][0]
                self.assertIsInstance(r, datetime)
                self.assertEqual(r, d)
                d = datetime(2016, 3, 14, 15, 9, 26, tzinfo=tzinfo)
                q = "select $1::timestamptz"
                r = query(q, (d,)).getresult()[0][0]
                self.assertIsInstance(r, datetime)
                self.assertEqual(r, d)
                d = datetime(2016, 3, 14, 15, 9, 26, 535897, tzinfo)
                q = "select $1::timestamptz"
                r = query(q, (d,)).getresult()[0][0]
                self.assertIsInstance(r, datetime)
                self.assertEqual(r, d)
                q = ("select '10000-08-01 AD'::timestamptz,"
                    " '0099-01-08 BC'::timestamptz")
                r = query(q).getresult()[0]
                self.assertIsInstance(r[0], datetime)
                self.assertIsInstance(r[1], datetime)
                self.assertEqual(r[0], datetime.max)
                self.assertEqual(r[1], datetime.min)
        q = "select 'infinity'::timestamptz, '-infinity'::timestamptz"
        r = query(q).getresult()[0]
        self.assertIsInstance(r[0], datetime)
        self.assertIsInstance(r[1], datetime)
        self.assertEqual(r[0], datetime.max)
        self.assertEqual(r[1], datetime.min)
项目:slack-sql    作者:wang502    | 项目源码 | 文件源码
def cast_timestamptz(value, connection):
    """Cast a timestamptz value."""
    if value == '-infinity':
        return datetime.min
    if value == 'infinity':
        return datetime.max
    value = value.split()
    if value[-1] == 'BC':
        return datetime.min
    fmt = connection.date_format()
    if fmt.endswith('-%Y') and len(value) > 2:
        value = value[1:]
        if len(value[3]) > 4:
            return datetime.max
        fmt = ['%d %b' if fmt.startswith('%d') else '%b %d',
            '%H:%M:%S.%f' if len(value[2]) > 8 else '%H:%M:%S', '%Y']
        value, tz = value[:-1], value[-1]
    else:
        if fmt.startswith('%Y-'):
            tz = _re_timezone.match(value[1])
            if tz:
                value[1], tz = tz.groups()
            else:
                tz = '+0000'
        else:
            value, tz = value[:-1], value[-1]
        if len(value[0]) > 10:
            return datetime.max
        fmt = [fmt, '%H:%M:%S.%f' if len(value[1]) > 8 else '%H:%M:%S']
    if _has_timezone:
        value.append(_timezone_as_offset(tz))
        fmt.append('%z')
        return datetime.strptime(' '.join(value), ' '.join(fmt))
    return datetime.strptime(' '.join(value), ' '.join(fmt)).replace(
        tzinfo=_get_timezone(tz))
项目:slack-sql    作者:wang502    | 项目源码 | 文件源码
def cast_timestamptz(value, connection):
    """Cast a timestamptz value."""
    if value == '-infinity':
        return datetime.min
    if value == 'infinity':
        return datetime.max
    value = value.split()
    if value[-1] == 'BC':
        return datetime.min
    fmt = connection.date_format()
    if fmt.endswith('-%Y') and len(value) > 2:
        value = value[1:]
        if len(value[3]) > 4:
            return datetime.max
        fmt = ['%d %b' if fmt.startswith('%d') else '%b %d',
            '%H:%M:%S.%f' if len(value[2]) > 8 else '%H:%M:%S', '%Y']
        value, tz = value[:-1], value[-1]
    else:
        if fmt.startswith('%Y-'):
            tz = _re_timezone.match(value[1])
            if tz:
                value[1], tz = tz.groups()
            else:
                tz = '+0000'
        else:
            value, tz = value[:-1], value[-1]
        if len(value[0]) > 10:
            return datetime.max
        fmt = [fmt, '%H:%M:%S.%f' if len(value[1]) > 8 else '%H:%M:%S']
    if _has_timezone:
        value.append(_timezone_as_offset(tz))
        fmt.append('%z')
        return datetime.strptime(' '.join(value), ' '.join(fmt))
    return datetime.strptime(' '.join(value), ' '.join(fmt)).replace(
        tzinfo=_get_timezone(tz))
项目:cuny-bdif    作者:aristotle-tek    | 项目源码 | 文件源码
def __init__(self, batch_name, sd, mimetype_files=None):
        self.sd = sd
        self.batch = batch_name
        self.log_fp = None
        self.num_files = 0
        self.total_time = 0
        self.min_time = timedelta.max
        self.max_time = timedelta.min
        self.earliest_time = datetime.max
        self.latest_time = datetime.min
        self.queue = self.sd.get_obj('output_queue')
        self.domain = self.sd.get_obj('output_domain')
项目:appcompatprocessor    作者:mbevilacqua    | 项目源码 | 文件源码
def calculateID(self, file_name_fullpath):
        # Get the creation date for the first PersistenceItem in the audit (they will all be the same)
        instanceID = datetime.min
        tmp_instanceID = None

        try:
            file_object = loadFile(file_name_fullpath)
            root = ET.parse(file_object).getroot()
            file_object.close()
            reg_key = root.find('PersistenceItem')
            reg_modified = reg_key.get('created')
            try:
                tmp_instanceID = datetime.strptime(reg_modified, "%Y-%m-%dT%H:%M:%SZ")
            except ValueError as e:
                tmp_instanceID = datetime.max
                logger.warning("Invalid reg_modified date found!: %s (%s)" % (reg_modified, file_name_fullpath))
            instanceID = tmp_instanceID
        except Exception:
            traceback.print_exc(file=sys.stdout)

        # If we found no PersistenceItem date we go with plan B (but most probably this is corrupt and will fail later)
        if instanceID is None:
            file_object = loadFile(file_name_fullpath)
            content = file_object.read()
            instanceID = hashlib.md5(content).hexdigest()
            file_object.close()

        return instanceID
项目:appcompatprocessor    作者:mbevilacqua    | 项目源码 | 文件源码
def calculateID(self, file_name_fullpath):
        # Get the creation date for the first PersistenceItem in the audit (they will all be the same)
        instanceID = datetime.min
        tmp_instanceID = None

        try:
            file_object = loadFile(file_name_fullpath)
            root = ET.parse(file_object).getroot()
            file_object.close()
            reg_key = root.find('AppCompatItemExtended')
            reg_modified = reg_key.get('created')
            try:
                tmp_instanceID = datetime.strptime(reg_modified, "%Y-%m-%dT%H:%M:%SZ")
            except ValueError as e:
                tmp_instanceID = datetime.max
                logger.warning("Invalid reg_modified date found!: %s (%s)" % (reg_modified, file_name_fullpath))
            instanceID = tmp_instanceID
        except Exception:
            traceback.print_exc(file=sys.stdout)

        # If we found no PersistenceItem date we go with plan B (but most probably this is corrupt and will fail later)
        if instanceID is None:
            file_object = loadFile(file_name_fullpath)
            content = file_object.read()
            instanceID = hashlib.md5(content).hexdigest()
            file_object.close()

        return instanceID
项目:appcompatprocessor    作者:mbevilacqua    | 项目源码 | 文件源码
def calculateID(self, file_name_fullpath):
        instanceID = datetime.min
        tmp_instanceID = None

        try:
            file_object = loadFile(file_name_fullpath)
            root = ET.parse(file_object).getroot()
            file_object.close()
            for reg_key in root.findall('RegistryItem'):
                tmp_reg_key = reg_key.find('Modified')
                if tmp_reg_key is not None:
                    reg_modified = tmp_reg_key.text
                    try:
                        tmp_instanceID = datetime.strptime(reg_modified, "%Y-%m-%dT%H:%M:%SZ")
                    except ValueError as e:
                        tmp_instanceID = datetime.max
                        logger.warning("Invalid reg_modified date found!: %s (%s)" % (reg_modified, file_name_fullpath))
                    if instanceID < tmp_instanceID:
                        instanceID = tmp_instanceID
                else:
                    logger.warning("Found RegistryItem with no Modified date (Mir bug?): %s" % file_name_fullpath)
        except Exception:
            logger.exception("Error on calculateID for: %s" % file_name_fullpath)

        # If we found no Modified date in any of the RegistryItems we go with plan B (but most probably ShimCacheParser will fail to parse anyway)
        if instanceID is None:
            file_object = loadFile(file_name_fullpath)
            content = file_object.read()
            instanceID = hashlib.md5(content).hexdigest()
            file_object.close()

        return instanceID