Wala shell

Advanced PHP File Manager

Server IP: 127.0.0.1 | PHP Version: 8.3.30 | Domains Hosted: 54

Name	Size	Last Modified	Actions
📁 ..	-	2026-03-21 15:45
📄 __init__.py """ future.backports package """ from __future__ import absolute_import import sys __future_module__ = True from future.standard_library import import_top_level_modules if sys.version_info[0] >= 3: import_top_level_modules() from .misc import (ceil, OrderedDict, Counter, ChainMap, check_output, count, recursive_repr, _count_elements, cmp_to_key )	530 B	2026-01-20 12:58
📁 __pycache__	-	2026-03-21 15:45
📄 _markupbase.py """Shared support for scanning document type declarations in HTML and XHTML. Backported for python-future from Python 3.3. Reason: ParserBase is an old-style class in the Python 2.7 source of markupbase.py, which I suspect might be the cause of sporadic unit-test failures on travis-ci.org with test_htmlparser.py. The test failures look like this: ====================================================================== ERROR: test_attr_entity_replacement (future.tests.test_htmlparser.AttributesStrictTestCase) ---------------------------------------------------------------------- Traceback (most recent call last): File "/home/travis/build/edschofield/python-future/future/tests/test_htmlparser.py", line 661, in test_attr_entity_replacement [("starttag", "a", [("b", "&><\"'")])]) File "/home/travis/build/edschofield/python-future/future/tests/test_htmlparser.py", line 93, in _run_check collector = self.get_collector() File "/home/travis/build/edschofield/python-future/future/tests/test_htmlparser.py", line 617, in get_collector return EventCollector(strict=True) File "/home/travis/build/edschofield/python-future/future/tests/test_htmlparser.py", line 27, in __init__ html.parser.HTMLParser.__init__(self, args, kw) File "/home/travis/build/edschofield/python-future/future/backports/html/parser.py", line 135, in __init__ self.reset() File "/home/travis/build/edschofield/python-future/future/backports/html/parser.py", line 143, in reset _markupbase.ParserBase.reset(self) TypeError: unbound method reset() must be called with ParserBase instance as first argument (got EventCollector instance instead) This module is used as a foundation for the html.parser module. It has no documented public API and should not be used directly. """ import re _declname_match = re.compile(r'[a-zA-Z][-_.a-zA-Z0-9]\s').match _declstringlit_match = re.compile(r'(\'[^\']\'\|"[^"]")\s').match _commentclose = re.compile(r'--\s>') _markedsectionclose = re.compile(r']\s]\s>') # An analysis of the MS-Word extensions is available at # http://www.planetpublish.com/xmlarena/xap/Thursday/WordtoXML.pdf _msmarkedsectionclose = re.compile(r']\s>') del re class ParserBase(object): """Parser base class which provides some common support methods used by the SGML/HTML and XHTML parsers.""" def __init__(self): if self.__class__ is ParserBase: raise RuntimeError( "_markupbase.ParserBase must be subclassed") def error(self, message): raise NotImplementedError( "subclasses of ParserBase must override error()") def reset(self): self.lineno = 1 self.offset = 0 def getpos(self): """Return current line number and offset.""" return self.lineno, self.offset # Internal -- update line number and offset. This should be # called for each piece of data exactly once, in order -- in other # words the concatenation of all the input strings to this # function should be exactly the entire input. def updatepos(self, i, j): if i >= j: return j rawdata = self.rawdata nlines = rawdata.count("\n", i, j) if nlines: self.lineno = self.lineno + nlines pos = rawdata.rindex("\n", i, j) # Should not fail self.offset = j-(pos+1) else: self.offset = self.offset + j-i return j _decl_otherchars = '' # Internal -- parse declaration (for use by subclasses). def parse_declaration(self, i): # This is some sort of declaration; in "HTML as # deployed," this should only be the document type # declaration ("<!DOCTYPE html...>"). # ISO 8879:1986, however, has more complex # declaration syntax for elements in <!...>, including: # --comment-- # [marked section] # name in the following list: ENTITY, DOCTYPE, ELEMENT, # ATTLIST, NOTATION, SHORTREF, USEMAP, # LINKTYPE, LINK, IDLINK, USELINK, SYSTEM rawdata = self.rawdata j = i + 2 assert rawdata[i:j] == "<!", "unexpected call to parse_declaration" if rawdata[j:j+1] == ">": # the empty comment <!> return j + 1 if rawdata[j:j+1] in ("-", ""): # Start of comment followed by buffer boundary, # or just a buffer boundary. return -1 # A simple, practical version could look like: ((name\|stringlit) S) + '>' n = len(rawdata) if rawdata[j:j+2] == '--': #comment # Locate --.-- as the body of the comment return self.parse_comment(i) elif rawdata[j] == '[': #marked section # Locate [statusWord [...arbitrary SGML...]] as the body of the marked section # Where statusWord is one of TEMP, CDATA, IGNORE, INCLUDE, RCDATA # Note that this is extended by Microsoft Office "Save as Web" function # to include [if...] and [endif]. return self.parse_marked_section(i) else: #all other declaration elements decltype, j = self._scan_name(j, i) if j < 0: return j if decltype == "doctype": self._decl_otherchars = '' while j < n: c = rawdata[j] if c == ">": # end of declaration syntax data = rawdata[i+2:j] if decltype == "doctype": self.handle_decl(data) else: # According to the HTML5 specs sections "8.2.4.44 Bogus # comment state" and "8.2.4.45 Markup declaration open # state", a comment token should be emitted. # Calling unknown_decl provides more flexibility though. self.unknown_decl(data) return j + 1 if c in "\"'": m = _declstringlit_match(rawdata, j) if not m: return -1 # incomplete j = m.end() elif c in "abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ": name, j = self._scan_name(j, i) elif c in self._decl_otherchars: j = j + 1 elif c == "[": # this could be handled in a separate doctype parser if decltype == "doctype": j = self._parse_doctype_subset(j + 1, i) elif decltype in set(["attlist", "linktype", "link", "element"]): # must tolerate []'d groups in a content model in an element declaration # also in data attribute specifications of attlist declaration # also link type declaration subsets in linktype declarations # also link attribute specification lists in link declarations self.error("unsupported '[' char in %s declaration" % decltype) else: self.error("unexpected '[' char in declaration") else: self.error( "unexpected %r char in declaration" % rawdata[j]) if j < 0: return j return -1 # incomplete # Internal -- parse a marked section # Override this to handle MS-word extension syntax <![if word]>content<![endif]> def parse_marked_section(self, i, report=1): rawdata= self.rawdata assert rawdata[i:i+3] == '<![', "unexpected call to parse_marked_section()" sectName, j = self._scan_name( i+3, i ) if j < 0: return j if sectName in set(["temp", "cdata", "ignore", "include", "rcdata"]): # look for standard ]]> ending match= _markedsectionclose.search(rawdata, i+3) elif sectName in set(["if", "else", "endif"]): # look for MS Office ]> ending match= _msmarkedsectionclose.search(rawdata, i+3) else: self.error('unknown status keyword %r in marked section' % rawdata[i+3:j]) if not match: return -1 if report: j = match.start(0) self.unknown_decl(rawdata[i+3: j]) return match.end(0) # Internal -- parse comment, return length or -1 if not terminated def parse_comment(self, i, report=1): rawdata = self.rawdata if rawdata[i:i+4] != '<!--': self.error('unexpected call to parse_comment()') match = _commentclose.search(rawdata, i+4) if not match: return -1 if report: j = match.start(0) self.handle_comment(rawdata[i+4: j]) return match.end(0) # Internal -- scan past the internal subset in a <!DOCTYPE declaration, # returning the index just past any whitespace following the trailing ']'. def _parse_doctype_subset(self, i, declstartpos): rawdata = self.rawdata n = len(rawdata) j = i while j < n: c = rawdata[j] if c == "<": s = rawdata[j:j+2] if s == "<": # end of buffer; incomplete return -1 if s != "<!": self.updatepos(declstartpos, j + 1) self.error("unexpected char in internal subset (in %r)" % s) if (j + 2) == n: # end of buffer; incomplete return -1 if (j + 4) > n: # end of buffer; incomplete return -1 if rawdata[j:j+4] == "<!--": j = self.parse_comment(j, report=0) if j < 0: return j continue name, j = self._scan_name(j + 2, declstartpos) if j == -1: return -1 if name not in set(["attlist", "element", "entity", "notation"]): self.updatepos(declstartpos, j + 2) self.error( "unknown declaration %r in internal subset" % name) # handle the individual names meth = getattr(self, "_parse_doctype_" + name) j = meth(j, declstartpos) if j < 0: return j elif c == "%": # parameter entity reference if (j + 1) == n: # end of buffer; incomplete return -1 s, j = self._scan_name(j + 1, declstartpos) if j < 0: return j if rawdata[j] == ";": j = j + 1 elif c == "]": j = j + 1 while j < n and rawdata[j].isspace(): j = j + 1 if j < n: if rawdata[j] == ">": return j self.updatepos(declstartpos, j) self.error("unexpected char after internal subset") else: return -1 elif c.isspace(): j = j + 1 else: self.updatepos(declstartpos, j) self.error("unexpected char %r in internal subset" % c) # end of buffer reached return -1 # Internal -- scan past <!ELEMENT declarations def _parse_doctype_element(self, i, declstartpos): name, j = self._scan_name(i, declstartpos) if j == -1: return -1 # style content model; just skip until '>' rawdata = self.rawdata if '>' in rawdata[j:]: return rawdata.find(">", j) + 1 return -1 # Internal -- scan past <!ATTLIST declarations def _parse_doctype_attlist(self, i, declstartpos): rawdata = self.rawdata name, j = self._scan_name(i, declstartpos) c = rawdata[j:j+1] if c == "": return -1 if c == ">": return j + 1 while 1: # scan a series of attribute descriptions; simplified: # name type [value] [#constraint] name, j = self._scan_name(j, declstartpos) if j < 0: return j c = rawdata[j:j+1] if c == "": return -1 if c == "(": # an enumerated type; look for ')' if ")" in rawdata[j:]: j = rawdata.find(")", j) + 1 else: return -1 while rawdata[j:j+1].isspace(): j = j + 1 if not rawdata[j:]: # end of buffer, incomplete return -1 else: name, j = self._scan_name(j, declstartpos) c = rawdata[j:j+1] if not c: return -1 if c in "'\"": m = _declstringlit_match(rawdata, j) if m: j = m.end() else: return -1 c = rawdata[j:j+1] if not c: return -1 if c == "#": if rawdata[j:] == "#": # end of buffer return -1 name, j = self._scan_name(j + 1, declstartpos) if j < 0: return j c = rawdata[j:j+1] if not c: return -1 if c == '>': # all done return j + 1 # Internal -- scan past <!NOTATION declarations def _parse_doctype_notation(self, i, declstartpos): name, j = self._scan_name(i, declstartpos) if j < 0: return j rawdata = self.rawdata while 1: c = rawdata[j:j+1] if not c: # end of buffer; incomplete return -1 if c == '>': return j + 1 if c in "'\"": m = _declstringlit_match(rawdata, j) if not m: return -1 j = m.end() else: name, j = self._scan_name(j, declstartpos) if j < 0: return j # Internal -- scan past <!ENTITY declarations def _parse_doctype_entity(self, i, declstartpos): rawdata = self.rawdata if rawdata[i:i+1] == "%": j = i + 1 while 1: c = rawdata[j:j+1] if not c: return -1 if c.isspace(): j = j + 1 else: break else: j = i name, j = self._scan_name(j, declstartpos) if j < 0: return j while 1: c = self.rawdata[j:j+1] if not c: return -1 if c in "'\"": m = _declstringlit_match(rawdata, j) if m: j = m.end() else: return -1 # incomplete elif c == ">": return j + 1 else: name, j = self._scan_name(j, declstartpos) if j < 0: return j # Internal -- scan a name token and the new position and the token, or # return -1 if we've reached the end of the buffer. def _scan_name(self, i, declstartpos): rawdata = self.rawdata n = len(rawdata) if i == n: return None, -1 m = _declname_match(rawdata, i) if m: s = m.group() name = s.strip() if (i + len(s)) == n: return None, -1 # end of buffer return name.lower(), m.end() else: self.updatepos(declstartpos, i) self.error("expected name token at %r" % rawdata[declstartpos:declstartpos+20]) # To be overridden -- handlers for unknown objects def unknown_decl(self, data): pass	15.83 KB	2026-01-20 12:58
📄 datetime.py """Concrete date/time and related types. See http://www.iana.org/time-zones/repository/tz-link.html for time zone and DST data sources. """ from __future__ import division from __future__ import unicode_literals from __future__ import print_function from __future__ import absolute_import from future.builtins import str from future.builtins import bytes from future.builtins import map from future.builtins import round from future.builtins import int from future.builtins import object from future.utils import native_str, PY2 import time as _time import math as _math def _cmp(x, y): return 0 if x == y else 1 if x > y else -1 MINYEAR = 1 MAXYEAR = 9999 _MAXORDINAL = 3652059 # date.max.toordinal() # Utility functions, adapted from Python's Demo/classes/Dates.py, which # also assumes the current Gregorian calendar indefinitely extended in # both directions. Difference: Dates.py calls January 1 of year 0 day # number 1. The code here calls January 1 of year 1 day number 1. This is # to match the definition of the "proleptic Gregorian" calendar in Dershowitz # and Reingold's "Calendrical Calculations", where it's the base calendar # for all computations. See the book for algorithms for converting between # proleptic Gregorian ordinals and many other calendar systems. _DAYS_IN_MONTH = [None, 31, 28, 31, 30, 31, 30, 31, 31, 30, 31, 30, 31] _DAYS_BEFORE_MONTH = [None] dbm = 0 for dim in _DAYS_IN_MONTH[1:]: _DAYS_BEFORE_MONTH.append(dbm) dbm += dim del dbm, dim def _is_leap(year): "year -> 1 if leap year, else 0." return year % 4 == 0 and (year % 100 != 0 or year % 400 == 0) def _days_before_year(year): "year -> number of days before January 1st of year." y = year - 1 return y365 + y//4 - y//100 + y//400 def _days_in_month(year, month): "year, month -> number of days in that month in that year." assert 1 <= month <= 12, month if month == 2 and _is_leap(year): return 29 return _DAYS_IN_MONTH[month] def _days_before_month(year, month): "year, month -> number of days in year preceding first day of month." assert 1 <= month <= 12, 'month must be in 1..12' return _DAYS_BEFORE_MONTH[month] + (month > 2 and _is_leap(year)) def _ymd2ord(year, month, day): "year, month, day -> ordinal, considering 01-Jan-0001 as day 1." assert 1 <= month <= 12, 'month must be in 1..12' dim = _days_in_month(year, month) assert 1 <= day <= dim, ('day must be in 1..%d' % dim) return (_days_before_year(year) + _days_before_month(year, month) + day) _DI400Y = _days_before_year(401) # number of days in 400 years _DI100Y = _days_before_year(101) # " " " " 100 " _DI4Y = _days_before_year(5) # " " " " 4 " # A 4-year cycle has an extra leap day over what we'd get from pasting # together 4 single years. assert _DI4Y == 4 365 + 1 # Similarly, a 400-year cycle has an extra leap day over what we'd get from # pasting together 4 100-year cycles. assert _DI400Y == 4 * _DI100Y + 1 # OTOH, a 100-year cycle has one fewer leap day than we'd get from # pasting together 25 4-year cycles. assert _DI100Y == 25 * _DI4Y - 1 def _ord2ymd(n): "ordinal -> (year, month, day), considering 01-Jan-0001 as day 1." # n is a 1-based index, starting at 1-Jan-1. The pattern of leap years # repeats exactly every 400 years. The basic strategy is to find the # closest 400-year boundary at or before n, then work with the offset # from that boundary to n. Life is much clearer if we subtract 1 from # n first -- then the values of n at 400-year boundaries are exactly # those divisible by _DI400Y: # # D M Y n n-1 # -- --- ---- ---------- ---------------- # 31 Dec -400 -_DI400Y -_DI400Y -1 # 1 Jan -399 -_DI400Y +1 -_DI400Y 400-year boundary # ... # 30 Dec 000 -1 -2 # 31 Dec 000 0 -1 # 1 Jan 001 1 0 400-year boundary # 2 Jan 001 2 1 # 3 Jan 001 3 2 # ... # 31 Dec 400 _DI400Y _DI400Y -1 # 1 Jan 401 _DI400Y +1 _DI400Y 400-year boundary n -= 1 n400, n = divmod(n, _DI400Y) year = n400 * 400 + 1 # ..., -399, 1, 401, ... # Now n is the (non-negative) offset, in days, from January 1 of year, to # the desired date. Now compute how many 100-year cycles precede n. # Note that it's possible for n100 to equal 4! In that case 4 full # 100-year cycles precede the desired day, which implies the desired # day is December 31 at the end of a 400-year cycle. n100, n = divmod(n, _DI100Y) # Now compute how many 4-year cycles precede it. n4, n = divmod(n, _DI4Y) # And now how many single years. Again n1 can be 4, and again meaning # that the desired day is December 31 at the end of the 4-year cycle. n1, n = divmod(n, 365) year += n100 * 100 + n4 * 4 + n1 if n1 == 4 or n100 == 4: assert n == 0 return year-1, 12, 31 # Now the year is correct, and n is the offset from January 1. We find # the month via an estimate that's either exact or one too large. leapyear = n1 == 3 and (n4 != 24 or n100 == 3) assert leapyear == _is_leap(year) month = (n + 50) >> 5 preceding = _DAYS_BEFORE_MONTH[month] + (month > 2 and leapyear) if preceding > n: # estimate is too large month -= 1 preceding -= _DAYS_IN_MONTH[month] + (month == 2 and leapyear) n -= preceding assert 0 <= n < _days_in_month(year, month) # Now the year and month are correct, and n is the offset from the # start of that month: we're done! return year, month, n+1 # Month and day names. For localized versions, see the calendar module. _MONTHNAMES = [None, "Jan", "Feb", "Mar", "Apr", "May", "Jun", "Jul", "Aug", "Sep", "Oct", "Nov", "Dec"] _DAYNAMES = [None, "Mon", "Tue", "Wed", "Thu", "Fri", "Sat", "Sun"] def _build_struct_time(y, m, d, hh, mm, ss, dstflag): wday = (_ymd2ord(y, m, d) + 6) % 7 dnum = _days_before_month(y, m) + d return _time.struct_time((y, m, d, hh, mm, ss, wday, dnum, dstflag)) def _format_time(hh, mm, ss, us): # Skip trailing microseconds when us==0. result = "%02d:%02d:%02d" % (hh, mm, ss) if us: result += ".%06d" % us return result # Correctly substitute for %z and %Z escapes in strftime formats. def _wrap_strftime(object, format, timetuple): # Don't call utcoffset() or tzname() unless actually needed. freplace = None # the string to use for %f zreplace = None # the string to use for %z Zreplace = None # the string to use for %Z # Scan format for %z and %Z escapes, replacing as needed. newformat = [] push = newformat.append i, n = 0, len(format) while i < n: ch = format[i] i += 1 if ch == '%': if i < n: ch = format[i] i += 1 if ch == 'f': if freplace is None: freplace = '%06d' % getattr(object, 'microsecond', 0) newformat.append(freplace) elif ch == 'z': if zreplace is None: zreplace = "" if hasattr(object, "utcoffset"): offset = object.utcoffset() if offset is not None: sign = '+' if offset.days < 0: offset = -offset sign = '-' h, m = divmod(offset, timedelta(hours=1)) assert not m % timedelta(minutes=1), "whole minute" m //= timedelta(minutes=1) zreplace = '%c%02d%02d' % (sign, h, m) assert '%' not in zreplace newformat.append(zreplace) elif ch == 'Z': if Zreplace is None: Zreplace = "" if hasattr(object, "tzname"): s = object.tzname() if s is not None: # strftime is going to have at this: escape % Zreplace = s.replace('%', '%%') newformat.append(Zreplace) else: push('%') push(ch) else: push('%') else: push(ch) newformat = "".join(newformat) return _time.strftime(newformat, timetuple) def _call_tzinfo_method(tzinfo, methname, tzinfoarg): if tzinfo is None: return None return getattr(tzinfo, methname)(tzinfoarg) # Just raise TypeError if the arg isn't None or a string. def _check_tzname(name): if name is not None and not isinstance(name, str): raise TypeError("tzinfo.tzname() must return None or string, " "not '%s'" % type(name)) # name is the offset-producing method, "utcoffset" or "dst". # offset is what it returned. # If offset isn't None or timedelta, raises TypeError. # If offset is None, returns None. # Else offset is checked for being in range, and a whole # of minutes. # If it is, its integer value is returned. Else ValueError is raised. def _check_utc_offset(name, offset): assert name in ("utcoffset", "dst") if offset is None: return if not isinstance(offset, timedelta): raise TypeError("tzinfo.%s() must return None " "or timedelta, not '%s'" % (name, type(offset))) if offset % timedelta(minutes=1) or offset.microseconds: raise ValueError("tzinfo.%s() must return a whole number " "of minutes, got %s" % (name, offset)) if not -timedelta(1) < offset < timedelta(1): raise ValueError("%s()=%s, must be must be strictly between" " -timedelta(hours=24) and timedelta(hours=24)" % (name, offset)) def _check_date_fields(year, month, day): if not isinstance(year, int): raise TypeError('int expected') if not MINYEAR <= year <= MAXYEAR: raise ValueError('year must be in %d..%d' % (MINYEAR, MAXYEAR), year) if not 1 <= month <= 12: raise ValueError('month must be in 1..12', month) dim = _days_in_month(year, month) if not 1 <= day <= dim: raise ValueError('day must be in 1..%d' % dim, day) def _check_time_fields(hour, minute, second, microsecond): if not isinstance(hour, int): raise TypeError('int expected') if not 0 <= hour <= 23: raise ValueError('hour must be in 0..23', hour) if not 0 <= minute <= 59: raise ValueError('minute must be in 0..59', minute) if not 0 <= second <= 59: raise ValueError('second must be in 0..59', second) if not 0 <= microsecond <= 999999: raise ValueError('microsecond must be in 0..999999', microsecond) def _check_tzinfo_arg(tz): if tz is not None and not isinstance(tz, tzinfo): raise TypeError("tzinfo argument must be None or of a tzinfo subclass") def _cmperror(x, y): raise TypeError("can't compare '%s' to '%s'" % ( type(x).__name__, type(y).__name__)) class timedelta(object): """Represent the difference between two datetime objects. Supported operators: - add, subtract timedelta - unary plus, minus, abs - compare to timedelta - multiply, divide by int In addition, datetime supports subtraction of two datetime objects returning a timedelta, and addition or subtraction of a datetime and a timedelta giving a datetime. Representation: (days, seconds, microseconds). Why? Because I felt like it. """ __slots__ = '_days', '_seconds', '_microseconds' def __new__(cls, days=0, seconds=0, microseconds=0, milliseconds=0, minutes=0, hours=0, weeks=0): # Doing this efficiently and accurately in C is going to be difficult # and error-prone, due to ubiquitous overflow possibilities, and that # C double doesn't have enough bits of precision to represent # microseconds over 10K years faithfully. The code here tries to make # explicit where go-fast assumptions can be relied on, in order to # guide the C implementation; it's way more convoluted than speed- # ignoring auto-overflow-to-long idiomatic Python could be. # XXX Check that all inputs are ints or floats. # Final values, all integer. # s and us fit in 32-bit signed ints; d isn't bounded. d = s = us = 0 # Normalize everything to days, seconds, microseconds. days += weeks7 seconds += minutes60 + hours3600 microseconds += milliseconds1000 # Get rid of all fractions, and normalize s and us. # Take a deep breath <wink>. if isinstance(days, float): dayfrac, days = _math.modf(days) daysecondsfrac, daysecondswhole = _math.modf(dayfrac * (24.3600.)) assert daysecondswhole == int(daysecondswhole) # can't overflow s = int(daysecondswhole) assert days == int(days) d = int(days) else: daysecondsfrac = 0.0 d = days assert isinstance(daysecondsfrac, float) assert abs(daysecondsfrac) <= 1.0 assert isinstance(d, int) assert abs(s) <= 24 3600 # days isn't referenced again before redefinition if isinstance(seconds, float): secondsfrac, seconds = _math.modf(seconds) assert seconds == int(seconds) seconds = int(seconds) secondsfrac += daysecondsfrac assert abs(secondsfrac) <= 2.0 else: secondsfrac = daysecondsfrac # daysecondsfrac isn't referenced again assert isinstance(secondsfrac, float) assert abs(secondsfrac) <= 2.0 assert isinstance(seconds, int) days, seconds = divmod(seconds, 243600) d += days s += int(seconds) # can't overflow assert isinstance(s, int) assert abs(s) <= 2 24 * 3600 # seconds isn't referenced again before redefinition usdouble = secondsfrac * 1e6 assert abs(usdouble) < 2.1e6 # exact value not critical # secondsfrac isn't referenced again if isinstance(microseconds, float): microseconds += usdouble microseconds = round(microseconds, 0) seconds, microseconds = divmod(microseconds, 1e6) assert microseconds == int(microseconds) assert seconds == int(seconds) days, seconds = divmod(seconds, 24.3600.) assert days == int(days) assert seconds == int(seconds) d += int(days) s += int(seconds) # can't overflow assert isinstance(s, int) assert abs(s) <= 3 24 * 3600 else: seconds, microseconds = divmod(microseconds, 1000000) days, seconds = divmod(seconds, 243600) d += days s += int(seconds) # can't overflow assert isinstance(s, int) assert abs(s) <= 3 24 * 3600 microseconds = float(microseconds) microseconds += usdouble microseconds = round(microseconds, 0) assert abs(s) <= 3 * 24 * 3600 assert abs(microseconds) < 3.1e6 # Just a little bit of carrying possible for microseconds and seconds. assert isinstance(microseconds, float) assert int(microseconds) == microseconds us = int(microseconds) seconds, us = divmod(us, 1000000) s += seconds # cant't overflow assert isinstance(s, int) days, s = divmod(s, 243600) d += days assert isinstance(d, int) assert isinstance(s, int) and 0 <= s < 243600 assert isinstance(us, int) and 0 <= us < 1000000 self = object.__new__(cls) self._days = d self._seconds = s self._microseconds = us if abs(d) > 999999999: raise OverflowError("timedelta # of days is too large: %d" % d) return self def __repr__(self): if self._microseconds: return "%s(%d, %d, %d)" % ('datetime.' + self.__class__.__name__, self._days, self._seconds, self._microseconds) if self._seconds: return "%s(%d, %d)" % ('datetime.' + self.__class__.__name__, self._days, self._seconds) return "%s(%d)" % ('datetime.' + self.__class__.__name__, self._days) def __str__(self): mm, ss = divmod(self._seconds, 60) hh, mm = divmod(mm, 60) s = "%d:%02d:%02d" % (hh, mm, ss) if self._days: def plural(n): return n, abs(n) != 1 and "s" or "" s = ("%d day%s, " % plural(self._days)) + s if self._microseconds: s = s + ".%06d" % self._microseconds return s def total_seconds(self): """Total seconds in the duration.""" return ((self.days * 86400 + self.seconds)106 + self.microseconds) / 106 # Read-only field accessors @property def days(self): """days""" return self._days @property def seconds(self): """seconds""" return self._seconds @property def microseconds(self): """microseconds""" return self._microseconds def __add__(self, other): if isinstance(other, timedelta): # for CPython compatibility, we cannot use # our __class__ here, but need a real timedelta return timedelta(self._days + other._days, self._seconds + other._seconds, self._microseconds + other._microseconds) return NotImplemented __radd__ = __add__ def __sub__(self, other): if isinstance(other, timedelta): # for CPython compatibility, we cannot use # our __class__ here, but need a real timedelta return timedelta(self._days - other._days, self._seconds - other._seconds, self._microseconds - other._microseconds) return NotImplemented def __rsub__(self, other): if isinstance(other, timedelta): return -self + other return NotImplemented def __neg__(self): # for CPython compatibility, we cannot use # our __class__ here, but need a real timedelta return timedelta(-self._days, -self._seconds, -self._microseconds) def __pos__(self): return self def __abs__(self): if self._days < 0: return -self else: return self def __mul__(self, other): if isinstance(other, int): # for CPython compatibility, we cannot use # our __class__ here, but need a real timedelta return timedelta(self._days other, self._seconds * other, self._microseconds * other) if isinstance(other, float): a, b = other.as_integer_ratio() return self * a / b return NotImplemented __rmul__ = __mul__ def _to_microseconds(self): return ((self._days * (243600) + self._seconds) 1000000 + self._microseconds) def __floordiv__(self, other): if not isinstance(other, (int, timedelta)): return NotImplemented usec = self._to_microseconds() if isinstance(other, timedelta): return usec // other._to_microseconds() if isinstance(other, int): return timedelta(0, 0, usec // other) def __truediv__(self, other): if not isinstance(other, (int, float, timedelta)): return NotImplemented usec = self._to_microseconds() if isinstance(other, timedelta): return usec / other._to_microseconds() if isinstance(other, int): return timedelta(0, 0, usec / other) if isinstance(other, float): a, b = other.as_integer_ratio() return timedelta(0, 0, b * usec / a) def __mod__(self, other): if isinstance(other, timedelta): r = self._to_microseconds() % other._to_microseconds() return timedelta(0, 0, r) return NotImplemented def __divmod__(self, other): if isinstance(other, timedelta): q, r = divmod(self._to_microseconds(), other._to_microseconds()) return q, timedelta(0, 0, r) return NotImplemented # Comparisons of timedelta objects with other. def __eq__(self, other): if isinstance(other, timedelta): return self._cmp(other) == 0 else: return False def __ne__(self, other): if isinstance(other, timedelta): return self._cmp(other) != 0 else: return True def __le__(self, other): if isinstance(other, timedelta): return self._cmp(other) <= 0 else: _cmperror(self, other) def __lt__(self, other): if isinstance(other, timedelta): return self._cmp(other) < 0 else: _cmperror(self, other) def __ge__(self, other): if isinstance(other, timedelta): return self._cmp(other) >= 0 else: _cmperror(self, other) def __gt__(self, other): if isinstance(other, timedelta): return self._cmp(other) > 0 else: _cmperror(self, other) def _cmp(self, other): assert isinstance(other, timedelta) return _cmp(self._getstate(), other._getstate()) def __hash__(self): return hash(self._getstate()) def __bool__(self): return (self._days != 0 or self._seconds != 0 or self._microseconds != 0) # Pickle support. def _getstate(self): return (self._days, self._seconds, self._microseconds) def __reduce__(self): return (self.__class__, self._getstate()) timedelta.min = timedelta(-999999999) timedelta.max = timedelta(days=999999999, hours=23, minutes=59, seconds=59, microseconds=999999) timedelta.resolution = timedelta(microseconds=1) class date(object): """Concrete date type. Constructors: __new__() fromtimestamp() today() fromordinal() Operators: __repr__, __str__ __cmp__, __hash__ __add__, __radd__, __sub__ (add/radd only with timedelta arg) Methods: timetuple() toordinal() weekday() isoweekday(), isocalendar(), isoformat() ctime() strftime() Properties (readonly): year, month, day """ __slots__ = '_year', '_month', '_day' def __new__(cls, year, month=None, day=None): """Constructor. Arguments: year, month, day (required, base 1) """ if (isinstance(year, bytes) and len(year) == 4 and 1 <= year[2] <= 12 and month is None): # Month is sane # Pickle support self = object.__new__(cls) self.__setstate(year) return self _check_date_fields(year, month, day) self = object.__new__(cls) self._year = year self._month = month self._day = day return self # Additional constructors @classmethod def fromtimestamp(cls, t): "Construct a date from a POSIX timestamp (like time.time())." y, m, d, hh, mm, ss, weekday, jday, dst = _time.localtime(t) return cls(y, m, d) @classmethod def today(cls): "Construct a date from time.time()." t = _time.time() return cls.fromtimestamp(t) @classmethod def fromordinal(cls, n): """Contruct a date from a proleptic Gregorian ordinal. January 1 of year 1 is day 1. Only the year, month and day are non-zero in the result. """ y, m, d = _ord2ymd(n) return cls(y, m, d) # Conversions to string def __repr__(self): """Convert to formal string, for repr(). >>> dt = datetime(2010, 1, 1) >>> repr(dt) 'datetime.datetime(2010, 1, 1, 0, 0)' >>> dt = datetime(2010, 1, 1, tzinfo=timezone.utc) >>> repr(dt) 'datetime.datetime(2010, 1, 1, 0, 0, tzinfo=datetime.timezone.utc)' """ return "%s(%d, %d, %d)" % ('datetime.' + self.__class__.__name__, self._year, self._month, self._day) # XXX These shouldn't depend on time.localtime(), because that # clips the usable dates to [1970 .. 2038). At least ctime() is # easily done without using strftime() -- that's better too because # strftime("%c", ...) is locale specific. def ctime(self): "Return ctime() style string." weekday = self.toordinal() % 7 or 7 return "%s %s %2d 00:00:00 %04d" % ( _DAYNAMES[weekday], _MONTHNAMES[self._month], self._day, self._year) def strftime(self, fmt): "Format using strftime()." return _wrap_strftime(self, fmt, self.timetuple()) def __format__(self, fmt): if len(fmt) != 0: return self.strftime(fmt) return str(self) def isoformat(self): """Return the date formatted according to ISO. This is 'YYYY-MM-DD'. References: - http://www.w3.org/TR/NOTE-datetime - http://www.cl.cam.ac.uk/~mgk25/iso-time.html """ return "%04d-%02d-%02d" % (self._year, self._month, self._day) __str__ = isoformat # Read-only field accessors @property def year(self): """year (1-9999)""" return self._year @property def month(self): """month (1-12)""" return self._month @property def day(self): """day (1-31)""" return self._day # Standard conversions, __cmp__, __hash__ (and helpers) def timetuple(self): "Return local time tuple compatible with time.localtime()." return _build_struct_time(self._year, self._month, self._day, 0, 0, 0, -1) def toordinal(self): """Return proleptic Gregorian ordinal for the year, month and day. January 1 of year 1 is day 1. Only the year, month and day values contribute to the result. """ return _ymd2ord(self._year, self._month, self._day) def replace(self, year=None, month=None, day=None): """Return a new date with new values for the specified fields.""" if year is None: year = self._year if month is None: month = self._month if day is None: day = self._day _check_date_fields(year, month, day) return date(year, month, day) # Comparisons of date objects with other. def __eq__(self, other): if isinstance(other, date): return self._cmp(other) == 0 return NotImplemented def __ne__(self, other): if isinstance(other, date): return self._cmp(other) != 0 return NotImplemented def __le__(self, other): if isinstance(other, date): return self._cmp(other) <= 0 return NotImplemented def __lt__(self, other): if isinstance(other, date): return self._cmp(other) < 0 return NotImplemented def __ge__(self, other): if isinstance(other, date): return self._cmp(other) >= 0 return NotImplemented def __gt__(self, other): if isinstance(other, date): return self._cmp(other) > 0 return NotImplemented def _cmp(self, other): assert isinstance(other, date) y, m, d = self._year, self._month, self._day y2, m2, d2 = other._year, other._month, other._day return _cmp((y, m, d), (y2, m2, d2)) def __hash__(self): "Hash." return hash(self._getstate()) # Computations def __add__(self, other): "Add a date to a timedelta." if isinstance(other, timedelta): o = self.toordinal() + other.days if 0 < o <= _MAXORDINAL: return date.fromordinal(o) raise OverflowError("result out of range") return NotImplemented __radd__ = __add__ def __sub__(self, other): """Subtract two dates, or a date and a timedelta.""" if isinstance(other, timedelta): return self + timedelta(-other.days) if isinstance(other, date): days1 = self.toordinal() days2 = other.toordinal() return timedelta(days1 - days2) return NotImplemented def weekday(self): "Return day of the week, where Monday == 0 ... Sunday == 6." return (self.toordinal() + 6) % 7 # Day-of-the-week and week-of-the-year, according to ISO def isoweekday(self): "Return day of the week, where Monday == 1 ... Sunday == 7." # 1-Jan-0001 is a Monday return self.toordinal() % 7 or 7 def isocalendar(self): """Return a 3-tuple containing ISO year, week number, and weekday. The first ISO week of the year is the (Mon-Sun) week containing the year's first Thursday; everything else derives from that. The first week is 1; Monday is 1 ... Sunday is 7. ISO calendar algorithm taken from http://www.phys.uu.nl/~vgent/calendar/isocalendar.htm """ year = self._year week1monday = _isoweek1monday(year) today = _ymd2ord(self._year, self._month, self._day) # Internally, week and day have origin 0 week, day = divmod(today - week1monday, 7) if week < 0: year -= 1 week1monday = _isoweek1monday(year) week, day = divmod(today - week1monday, 7) elif week >= 52: if today >= _isoweek1monday(year+1): year += 1 week = 0 return year, week+1, day+1 # Pickle support. def _getstate(self): yhi, ylo = divmod(self._year, 256) return bytes([yhi, ylo, self._month, self._day]), def __setstate(self, string): if len(string) != 4 or not (1 <= string[2] <= 12): raise TypeError("not enough arguments") yhi, ylo, self._month, self._day = string self._year = yhi * 256 + ylo def __reduce__(self): return (self.__class__, self._getstate()) _date_class = date # so functions w/ args named "date" can get at the class date.min = date(1, 1, 1) date.max = date(9999, 12, 31) date.resolution = timedelta(days=1) class tzinfo(object): """Abstract base class for time zone info classes. Subclasses must override the name(), utcoffset() and dst() methods. """ __slots__ = () def tzname(self, dt): "datetime -> string name of time zone." raise NotImplementedError("tzinfo subclass must override tzname()") def utcoffset(self, dt): "datetime -> minutes east of UTC (negative for west of UTC)" raise NotImplementedError("tzinfo subclass must override utcoffset()") def dst(self, dt): """datetime -> DST offset in minutes east of UTC. Return 0 if DST not in effect. utcoffset() must include the DST offset. """ raise NotImplementedError("tzinfo subclass must override dst()") def fromutc(self, dt): "datetime in UTC -> datetime in local time." if not isinstance(dt, datetime): raise TypeError("fromutc() requires a datetime argument") if dt.tzinfo is not self: raise ValueError("dt.tzinfo is not self") dtoff = dt.utcoffset() if dtoff is None: raise ValueError("fromutc() requires a non-None utcoffset() " "result") # See the long comment block at the end of this file for an # explanation of this algorithm. dtdst = dt.dst() if dtdst is None: raise ValueError("fromutc() requires a non-None dst() result") delta = dtoff - dtdst if delta: dt += delta dtdst = dt.dst() if dtdst is None: raise ValueError("fromutc(): dt.dst gave inconsistent " "results; cannot convert") return dt + dtdst # Pickle support. def __reduce__(self): getinitargs = getattr(self, "__getinitargs__", None) if getinitargs: args = getinitargs() else: args = () getstate = getattr(self, "__getstate__", None) if getstate: state = getstate() else: state = getattr(self, "__dict__", None) or None if state is None: return (self.__class__, args) else: return (self.__class__, args, state) _tzinfo_class = tzinfo class time(object): """Time with time zone. Constructors: __new__() Operators: __repr__, __str__ __cmp__, __hash__ Methods: strftime() isoformat() utcoffset() tzname() dst() Properties (readonly): hour, minute, second, microsecond, tzinfo """ def __new__(cls, hour=0, minute=0, second=0, microsecond=0, tzinfo=None): """Constructor. Arguments: hour, minute (required) second, microsecond (default to zero) tzinfo (default to None) """ self = object.__new__(cls) if isinstance(hour, bytes) and len(hour) == 6: # Pickle support self.__setstate(hour, minute or None) return self _check_tzinfo_arg(tzinfo) _check_time_fields(hour, minute, second, microsecond) self._hour = hour self._minute = minute self._second = second self._microsecond = microsecond self._tzinfo = tzinfo return self # Read-only field accessors @property def hour(self): """hour (0-23)""" return self._hour @property def minute(self): """minute (0-59)""" return self._minute @property def second(self): """second (0-59)""" return self._second @property def microsecond(self): """microsecond (0-999999)""" return self._microsecond @property def tzinfo(self): """timezone info object""" return self._tzinfo # Standard conversions, __hash__ (and helpers) # Comparisons of time objects with other. def __eq__(self, other): if isinstance(other, time): return self._cmp(other, allow_mixed=True) == 0 else: return False def __ne__(self, other): if isinstance(other, time): return self._cmp(other, allow_mixed=True) != 0 else: return True def __le__(self, other): if isinstance(other, time): return self._cmp(other) <= 0 else: _cmperror(self, other) def __lt__(self, other): if isinstance(other, time): return self._cmp(other) < 0 else: _cmperror(self, other) def __ge__(self, other): if isinstance(other, time): return self._cmp(other) >= 0 else: _cmperror(self, other) def __gt__(self, other): if isinstance(other, time): return self._cmp(other) > 0 else: _cmperror(self, other) def _cmp(self, other, allow_mixed=False): assert isinstance(other, time) mytz = self._tzinfo ottz = other._tzinfo myoff = otoff = None if mytz is ottz: base_compare = True else: myoff = self.utcoffset() otoff = other.utcoffset() base_compare = myoff == otoff if base_compare: return _cmp((self._hour, self._minute, self._second, self._microsecond), (other._hour, other._minute, other._second, other._microsecond)) if myoff is None or otoff is None: if allow_mixed: return 2 # arbitrary non-zero value else: raise TypeError("cannot compare naive and aware times") myhhmm = self._hour * 60 + self._minute - myoff//timedelta(minutes=1) othhmm = other._hour * 60 + other._minute - otoff//timedelta(minutes=1) return _cmp((myhhmm, self._second, self._microsecond), (othhmm, other._second, other._microsecond)) def __hash__(self): """Hash.""" tzoff = self.utcoffset() if not tzoff: # zero or None return hash(self._getstate()[0]) h, m = divmod(timedelta(hours=self.hour, minutes=self.minute) - tzoff, timedelta(hours=1)) assert not m % timedelta(minutes=1), "whole minute" m //= timedelta(minutes=1) if 0 <= h < 24: return hash(time(h, m, self.second, self.microsecond)) return hash((h, m, self.second, self.microsecond)) # Conversion to string def _tzstr(self, sep=":"): """Return formatted timezone offset (+xx:xx) or None.""" off = self.utcoffset() if off is not None: if off.days < 0: sign = "-" off = -off else: sign = "+" hh, mm = divmod(off, timedelta(hours=1)) assert not mm % timedelta(minutes=1), "whole minute" mm //= timedelta(minutes=1) assert 0 <= hh < 24 off = "%s%02d%s%02d" % (sign, hh, sep, mm) return off def __repr__(self): """Convert to formal string, for repr().""" if self._microsecond != 0: s = ", %d, %d" % (self._second, self._microsecond) elif self._second != 0: s = ", %d" % self._second else: s = "" s= "%s(%d, %d%s)" % ('datetime.' + self.__class__.__name__, self._hour, self._minute, s) if self._tzinfo is not None: assert s[-1:] == ")" s = s[:-1] + ", tzinfo=%r" % self._tzinfo + ")" return s def isoformat(self): """Return the time formatted according to ISO. This is 'HH:MM:SS.mmmmmm+zz:zz', or 'HH:MM:SS+zz:zz' if self.microsecond == 0. """ s = _format_time(self._hour, self._minute, self._second, self._microsecond) tz = self._tzstr() if tz: s += tz return s __str__ = isoformat def strftime(self, fmt): """Format using strftime(). The date part of the timestamp passed to underlying strftime should not be used. """ # The year must be >= 1000 else Python's strftime implementation # can raise a bogus exception. timetuple = (1900, 1, 1, self._hour, self._minute, self._second, 0, 1, -1) return _wrap_strftime(self, fmt, timetuple) def __format__(self, fmt): if len(fmt) != 0: return self.strftime(fmt) return str(self) # Timezone functions def utcoffset(self): """Return the timezone offset in minutes east of UTC (negative west of UTC).""" if self._tzinfo is None: return None offset = self._tzinfo.utcoffset(None) _check_utc_offset("utcoffset", offset) return offset def tzname(self): """Return the timezone name. Note that the name is 100% informational -- there's no requirement that it mean anything in particular. For example, "GMT", "UTC", "-500", "-5:00", "EDT", "US/Eastern", "America/New York" are all valid replies. """ if self._tzinfo is None: return None name = self._tzinfo.tzname(None) _check_tzname(name) return name def dst(self): """Return 0 if DST is not in effect, or the DST offset (in minutes eastward) if DST is in effect. This is purely informational; the DST offset has already been added to the UTC offset returned by utcoffset() if applicable, so there's no need to consult dst() unless you're interested in displaying the DST info. """ if self._tzinfo is None: return None offset = self._tzinfo.dst(None) _check_utc_offset("dst", offset) return offset def replace(self, hour=None, minute=None, second=None, microsecond=None, tzinfo=True): """Return a new time with new values for the specified fields.""" if hour is None: hour = self.hour if minute is None: minute = self.minute if second is None: second = self.second if microsecond is None: microsecond = self.microsecond if tzinfo is True: tzinfo = self.tzinfo _check_time_fields(hour, minute, second, microsecond) _check_tzinfo_arg(tzinfo) return time(hour, minute, second, microsecond, tzinfo) def __bool__(self): if self.second or self.microsecond: return True offset = self.utcoffset() or timedelta(0) return timedelta(hours=self.hour, minutes=self.minute) != offset # Pickle support. def _getstate(self): us2, us3 = divmod(self._microsecond, 256) us1, us2 = divmod(us2, 256) basestate = bytes([self._hour, self._minute, self._second, us1, us2, us3]) if self._tzinfo is None: return (basestate,) else: return (basestate, self._tzinfo) def __setstate(self, string, tzinfo): if len(string) != 6 or string[0] >= 24: raise TypeError("an integer is required") (self._hour, self._minute, self._second, us1, us2, us3) = string self._microsecond = (((us1 << 8) \| us2) << 8) \| us3 if tzinfo is None or isinstance(tzinfo, _tzinfo_class): self._tzinfo = tzinfo else: raise TypeError("bad tzinfo state arg %r" % tzinfo) def __reduce__(self): return (time, self._getstate()) _time_class = time # so functions w/ args named "time" can get at the class time.min = time(0, 0, 0) time.max = time(23, 59, 59, 999999) time.resolution = timedelta(microseconds=1) class datetime(date): """datetime(year, month, day[, hour[, minute[, second[, microsecond[,tzinfo]]]]]) The year, month and day arguments are required. tzinfo may be None, or an instance of a tzinfo subclass. The remaining arguments may be ints. """ __slots__ = date.__slots__ + ( '_hour', '_minute', '_second', '_microsecond', '_tzinfo') def __new__(cls, year, month=None, day=None, hour=0, minute=0, second=0, microsecond=0, tzinfo=None): if isinstance(year, bytes) and len(year) == 10: # Pickle support self = date.__new__(cls, year[:4]) self.__setstate(year, month) return self _check_tzinfo_arg(tzinfo) _check_time_fields(hour, minute, second, microsecond) self = date.__new__(cls, year, month, day) self._hour = hour self._minute = minute self._second = second self._microsecond = microsecond self._tzinfo = tzinfo return self # Read-only field accessors @property def hour(self): """hour (0-23)""" return self._hour @property def minute(self): """minute (0-59)""" return self._minute @property def second(self): """second (0-59)""" return self._second @property def microsecond(self): """microsecond (0-999999)""" return self._microsecond @property def tzinfo(self): """timezone info object""" return self._tzinfo @classmethod def fromtimestamp(cls, t, tz=None): """Construct a datetime from a POSIX timestamp (like time.time()). A timezone info object may be passed in as well. """ _check_tzinfo_arg(tz) converter = _time.localtime if tz is None else _time.gmtime t, frac = divmod(t, 1.0) us = int(frac * 1e6) # If timestamp is less than one microsecond smaller than a # full second, us can be rounded up to 1000000. In this case, # roll over to seconds, otherwise, ValueError is raised # by the constructor. if us == 1000000: t += 1 us = 0 y, m, d, hh, mm, ss, weekday, jday, dst = converter(t) ss = min(ss, 59) # clamp out leap seconds if the platform has them result = cls(y, m, d, hh, mm, ss, us, tz) if tz is not None: result = tz.fromutc(result) return result @classmethod def utcfromtimestamp(cls, t): "Construct a UTC datetime from a POSIX timestamp (like time.time())." t, frac = divmod(t, 1.0) us = int(frac * 1e6) # If timestamp is less than one microsecond smaller than a # full second, us can be rounded up to 1000000. In this case, # roll over to seconds, otherwise, ValueError is raised # by the constructor. if us == 1000000: t += 1 us = 0 y, m, d, hh, mm, ss, weekday, jday, dst = _time.gmtime(t) ss = min(ss, 59) # clamp out leap seconds if the platform has them return cls(y, m, d, hh, mm, ss, us) # XXX This is supposed to do better than we can do by using time.time(), # XXX if the platform supports a more accurate way. The C implementation # XXX uses gettimeofday on platforms that have it, but that isn't # XXX available from Python. So now() may return different results # XXX across the implementations. @classmethod def now(cls, tz=None): "Construct a datetime from time.time() and optional time zone info." t = _time.time() return cls.fromtimestamp(t, tz) @classmethod def utcnow(cls): "Construct a UTC datetime from time.time()." t = _time.time() return cls.utcfromtimestamp(t) @classmethod def combine(cls, date, time): "Construct a datetime from a given date and a given time." if not isinstance(date, _date_class): raise TypeError("date argument must be a date instance") if not isinstance(time, _time_class): raise TypeError("time argument must be a time instance") return cls(date.year, date.month, date.day, time.hour, time.minute, time.second, time.microsecond, time.tzinfo) def timetuple(self): "Return local time tuple compatible with time.localtime()." dst = self.dst() if dst is None: dst = -1 elif dst: dst = 1 else: dst = 0 return _build_struct_time(self.year, self.month, self.day, self.hour, self.minute, self.second, dst) def timestamp(self): "Return POSIX timestamp as float" if self._tzinfo is None: return _time.mktime((self.year, self.month, self.day, self.hour, self.minute, self.second, -1, -1, -1)) + self.microsecond / 1e6 else: return (self - _EPOCH).total_seconds() def utctimetuple(self): "Return UTC time tuple compatible with time.gmtime()." offset = self.utcoffset() if offset: self -= offset y, m, d = self.year, self.month, self.day hh, mm, ss = self.hour, self.minute, self.second return _build_struct_time(y, m, d, hh, mm, ss, 0) def date(self): "Return the date part." return date(self._year, self._month, self._day) def time(self): "Return the time part, with tzinfo None." return time(self.hour, self.minute, self.second, self.microsecond) def timetz(self): "Return the time part, with same tzinfo." return time(self.hour, self.minute, self.second, self.microsecond, self._tzinfo) def replace(self, year=None, month=None, day=None, hour=None, minute=None, second=None, microsecond=None, tzinfo=True): """Return a new datetime with new values for the specified fields.""" if year is None: year = self.year if month is None: month = self.month if day is None: day = self.day if hour is None: hour = self.hour if minute is None: minute = self.minute if second is None: second = self.second if microsecond is None: microsecond = self.microsecond if tzinfo is True: tzinfo = self.tzinfo _check_date_fields(year, month, day) _check_time_fields(hour, minute, second, microsecond) _check_tzinfo_arg(tzinfo) return datetime(year, month, day, hour, minute, second, microsecond, tzinfo) def astimezone(self, tz=None): if tz is None: if self.tzinfo is None: raise ValueError("astimezone() requires an aware datetime") ts = (self - _EPOCH) // timedelta(seconds=1) localtm = _time.localtime(ts) local = datetime(localtm[:6]) try: # Extract TZ data if available gmtoff = localtm.tm_gmtoff zone = localtm.tm_zone except AttributeError: # Compute UTC offset and compare with the value implied # by tm_isdst. If the values match, use the zone name # implied by tm_isdst. delta = local - datetime(_time.gmtime(ts)[:6]) dst = _time.daylight and localtm.tm_isdst > 0 gmtoff = -(_time.altzone if dst else _time.timezone) if delta == timedelta(seconds=gmtoff): tz = timezone(delta, _time.tzname[dst]) else: tz = timezone(delta) else: tz = timezone(timedelta(seconds=gmtoff), zone) elif not isinstance(tz, tzinfo): raise TypeError("tz argument must be an instance of tzinfo") mytz = self.tzinfo if mytz is None: raise ValueError("astimezone() requires an aware datetime") if tz is mytz: return self # Convert self to UTC, and attach the new time zone object. myoffset = self.utcoffset() if myoffset is None: raise ValueError("astimezone() requires an aware datetime") utc = (self - myoffset).replace(tzinfo=tz) # Convert from UTC to tz's local time. return tz.fromutc(utc) # Ways to produce a string. def ctime(self): "Return ctime() style string." weekday = self.toordinal() % 7 or 7 return "%s %s %2d %02d:%02d:%02d %04d" % ( _DAYNAMES[weekday], _MONTHNAMES[self._month], self._day, self._hour, self._minute, self._second, self._year) def isoformat(self, sep='T'): """Return the time formatted according to ISO. This is 'YYYY-MM-DD HH:MM:SS.mmmmmm', or 'YYYY-MM-DD HH:MM:SS' if self.microsecond == 0. If self.tzinfo is not None, the UTC offset is also attached, giving 'YYYY-MM-DD HH:MM:SS.mmmmmm+HH:MM' or 'YYYY-MM-DD HH:MM:SS+HH:MM'. Optional argument sep specifies the separator between date and time, default 'T'. """ s = ("%04d-%02d-%02d%c" % (self._year, self._month, self._day, sep) + _format_time(self._hour, self._minute, self._second, self._microsecond)) off = self.utcoffset() if off is not None: if off.days < 0: sign = "-" off = -off else: sign = "+" hh, mm = divmod(off, timedelta(hours=1)) assert not mm % timedelta(minutes=1), "whole minute" mm //= timedelta(minutes=1) s += "%s%02d:%02d" % (sign, hh, mm) return s def __repr__(self): """Convert to formal string, for repr().""" L = [self._year, self._month, self._day, # These are never zero self._hour, self._minute, self._second, self._microsecond] if L[-1] == 0: del L[-1] if L[-1] == 0: del L[-1] s = ", ".join(map(str, L)) s = "%s(%s)" % ('datetime.' + self.__class__.__name__, s) if self._tzinfo is not None: assert s[-1:] == ")" s = s[:-1] + ", tzinfo=%r" % self._tzinfo + ")" return s def __str__(self): "Convert to string, for str()." return self.isoformat(sep=' ') @classmethod def strptime(cls, date_string, format): 'string, format -> new datetime parsed from a string (like time.strptime()).' import _strptime return _strptime._strptime_datetime(cls, date_string, format) def utcoffset(self): """Return the timezone offset in minutes east of UTC (negative west of UTC).""" if self._tzinfo is None: return None offset = self._tzinfo.utcoffset(self) _check_utc_offset("utcoffset", offset) return offset def tzname(self): """Return the timezone name. Note that the name is 100% informational -- there's no requirement that it mean anything in particular. For example, "GMT", "UTC", "-500", "-5:00", "EDT", "US/Eastern", "America/New York" are all valid replies. """ name = _call_tzinfo_method(self._tzinfo, "tzname", self) _check_tzname(name) return name def dst(self): """Return 0 if DST is not in effect, or the DST offset (in minutes eastward) if DST is in effect. This is purely informational; the DST offset has already been added to the UTC offset returned by utcoffset() if applicable, so there's no need to consult dst() unless you're interested in displaying the DST info. """ if self._tzinfo is None: return None offset = self._tzinfo.dst(self) _check_utc_offset("dst", offset) return offset # Comparisons of datetime objects with other. def __eq__(self, other): if isinstance(other, datetime): return self._cmp(other, allow_mixed=True) == 0 elif not isinstance(other, date): return NotImplemented else: return False def __ne__(self, other): if isinstance(other, datetime): return self._cmp(other, allow_mixed=True) != 0 elif not isinstance(other, date): return NotImplemented else: return True def __le__(self, other): if isinstance(other, datetime): return self._cmp(other) <= 0 elif not isinstance(other, date): return NotImplemented else: _cmperror(self, other) def __lt__(self, other): if isinstance(other, datetime): return self._cmp(other) < 0 elif not isinstance(other, date): return NotImplemented else: _cmperror(self, other) def __ge__(self, other): if isinstance(other, datetime): return self._cmp(other) >= 0 elif not isinstance(other, date): return NotImplemented else: _cmperror(self, other) def __gt__(self, other): if isinstance(other, datetime): return self._cmp(other) > 0 elif not isinstance(other, date): return NotImplemented else: _cmperror(self, other) def _cmp(self, other, allow_mixed=False): assert isinstance(other, datetime) mytz = self._tzinfo ottz = other._tzinfo myoff = otoff = None if mytz is ottz: base_compare = True else: myoff = self.utcoffset() otoff = other.utcoffset() base_compare = myoff == otoff if base_compare: return _cmp((self._year, self._month, self._day, self._hour, self._minute, self._second, self._microsecond), (other._year, other._month, other._day, other._hour, other._minute, other._second, other._microsecond)) if myoff is None or otoff is None: if allow_mixed: return 2 # arbitrary non-zero value else: raise TypeError("cannot compare naive and aware datetimes") # XXX What follows could be done more efficiently... diff = self - other # this will take offsets into account if diff.days < 0: return -1 return diff and 1 or 0 def __add__(self, other): "Add a datetime and a timedelta." if not isinstance(other, timedelta): return NotImplemented delta = timedelta(self.toordinal(), hours=self._hour, minutes=self._minute, seconds=self._second, microseconds=self._microsecond) delta += other hour, rem = divmod(delta.seconds, 3600) minute, second = divmod(rem, 60) if 0 < delta.days <= _MAXORDINAL: return datetime.combine(date.fromordinal(delta.days), time(hour, minute, second, delta.microseconds, tzinfo=self._tzinfo)) raise OverflowError("result out of range") __radd__ = __add__ def __sub__(self, other): "Subtract two datetimes, or a datetime and a timedelta." if not isinstance(other, datetime): if isinstance(other, timedelta): return self + -other return NotImplemented days1 = self.toordinal() days2 = other.toordinal() secs1 = self._second + self._minute * 60 + self._hour * 3600 secs2 = other._second + other._minute * 60 + other._hour * 3600 base = timedelta(days1 - days2, secs1 - secs2, self._microsecond - other._microsecond) if self._tzinfo is other._tzinfo: return base myoff = self.utcoffset() otoff = other.utcoffset() if myoff == otoff: return base if myoff is None or otoff is None: raise TypeError("cannot mix naive and timezone-aware time") return base + otoff - myoff def __hash__(self): tzoff = self.utcoffset() if tzoff is None: return hash(self._getstate()[0]) days = _ymd2ord(self.year, self.month, self.day) seconds = self.hour * 3600 + self.minute * 60 + self.second return hash(timedelta(days, seconds, self.microsecond) - tzoff) # Pickle support. def _getstate(self): yhi, ylo = divmod(self._year, 256) us2, us3 = divmod(self._microsecond, 256) us1, us2 = divmod(us2, 256) basestate = bytes([yhi, ylo, self._month, self._day, self._hour, self._minute, self._second, us1, us2, us3]) if self._tzinfo is None: return (basestate,) else: return (basestate, self._tzinfo) def __setstate(self, string, tzinfo): (yhi, ylo, self._month, self._day, self._hour, self._minute, self._second, us1, us2, us3) = string self._year = yhi * 256 + ylo self._microsecond = (((us1 << 8) \| us2) << 8) \| us3 if tzinfo is None or isinstance(tzinfo, _tzinfo_class): self._tzinfo = tzinfo else: raise TypeError("bad tzinfo state arg %r" % tzinfo) def __reduce__(self): return (self.__class__, self._getstate()) datetime.min = datetime(1, 1, 1) datetime.max = datetime(9999, 12, 31, 23, 59, 59, 999999) datetime.resolution = timedelta(microseconds=1) def _isoweek1monday(year): # Helper to calculate the day number of the Monday starting week 1 # XXX This could be done more efficiently THURSDAY = 3 firstday = _ymd2ord(year, 1, 1) firstweekday = (firstday + 6) % 7 # See weekday() above week1monday = firstday - firstweekday if firstweekday > THURSDAY: week1monday += 7 return week1monday class timezone(tzinfo): __slots__ = '_offset', '_name' # Sentinel value to disallow None _Omitted = object() def __new__(cls, offset, name=_Omitted): if not isinstance(offset, timedelta): raise TypeError("offset must be a timedelta") if name is cls._Omitted: if not offset: return cls.utc name = None elif not isinstance(name, str): ### # For Python-Future: if PY2 and isinstance(name, native_str): name = name.decode() else: raise TypeError("name must be a string") ### if not cls._minoffset <= offset <= cls._maxoffset: raise ValueError("offset must be a timedelta" " strictly between -timedelta(hours=24) and" " timedelta(hours=24).") if (offset.microseconds != 0 or offset.seconds % 60 != 0): raise ValueError("offset must be a timedelta" " representing a whole number of minutes") return cls._create(offset, name) @classmethod def _create(cls, offset, name=None): self = tzinfo.__new__(cls) self._offset = offset self._name = name return self def __getinitargs__(self): """pickle support""" if self._name is None: return (self._offset,) return (self._offset, self._name) def __eq__(self, other): if type(other) != timezone: return False return self._offset == other._offset def __hash__(self): return hash(self._offset) def __repr__(self): """Convert to formal string, for repr(). >>> tz = timezone.utc >>> repr(tz) 'datetime.timezone.utc' >>> tz = timezone(timedelta(hours=-5), 'EST') >>> repr(tz) "datetime.timezone(datetime.timedelta(-1, 68400), 'EST')" """ if self is self.utc: return 'datetime.timezone.utc' if self._name is None: return "%s(%r)" % ('datetime.' + self.__class__.__name__, self._offset) return "%s(%r, %r)" % ('datetime.' + self.__class__.__name__, self._offset, self._name) def __str__(self): return self.tzname(None) def utcoffset(self, dt): if isinstance(dt, datetime) or dt is None: return self._offset raise TypeError("utcoffset() argument must be a datetime instance" " or None") def tzname(self, dt): if isinstance(dt, datetime) or dt is None: if self._name is None: return self._name_from_offset(self._offset) return self._name raise TypeError("tzname() argument must be a datetime instance" " or None") def dst(self, dt): if isinstance(dt, datetime) or dt is None: return None raise TypeError("dst() argument must be a datetime instance" " or None") def fromutc(self, dt): if isinstance(dt, datetime): if dt.tzinfo is not self: raise ValueError("fromutc: dt.tzinfo " "is not self") return dt + self._offset raise TypeError("fromutc() argument must be a datetime instance" " or None") _maxoffset = timedelta(hours=23, minutes=59) _minoffset = -_maxoffset @staticmethod def _name_from_offset(delta): if delta < timedelta(0): sign = '-' delta = -delta else: sign = '+' hours, rest = divmod(delta, timedelta(hours=1)) minutes = rest // timedelta(minutes=1) return 'UTC{}{:02d}:{:02d}'.format(sign, hours, minutes) timezone.utc = timezone._create(timedelta(0)) timezone.min = timezone._create(timezone._minoffset) timezone.max = timezone._create(timezone._maxoffset) _EPOCH = datetime(1970, 1, 1, tzinfo=timezone.utc) """ Some time zone algebra. For a datetime x, let x.n = x stripped of its timezone -- its naive time. x.o = x.utcoffset(), and assuming that doesn't raise an exception or return None x.d = x.dst(), and assuming that doesn't raise an exception or return None x.s = x's standard offset, x.o - x.d Now some derived rules, where k is a duration (timedelta). 1. x.o = x.s + x.d This follows from the definition of x.s. 2. If x and y have the same tzinfo member, x.s = y.s. This is actually a requirement, an assumption we need to make about sane tzinfo classes. 3. The naive UTC time corresponding to x is x.n - x.o. This is again a requirement for a sane tzinfo class. 4. (x+k).s = x.s This follows from #2, and that datimetimetz+timedelta preserves tzinfo. 5. (x+k).n = x.n + k Again follows from how arithmetic is defined. Now we can explain tz.fromutc(x). Let's assume it's an interesting case (meaning that the various tzinfo methods exist, and don't blow up or return None when called). The function wants to return a datetime y with timezone tz, equivalent to x. x is already in UTC. By #3, we want y.n - y.o = x.n [1] The algorithm starts by attaching tz to x.n, and calling that y. So x.n = y.n at the start. Then it wants to add a duration k to y, so that [1] becomes true; in effect, we want to solve [2] for k: (y+k).n - (y+k).o = x.n [2] By #1, this is the same as (y+k).n - ((y+k).s + (y+k).d) = x.n [3] By #5, (y+k).n = y.n + k, which equals x.n + k because x.n=y.n at the start. Substituting that into [3], x.n + k - (y+k).s - (y+k).d = x.n; the x.n terms cancel, leaving k - (y+k).s - (y+k).d = 0; rearranging, k = (y+k).s - (y+k).d; by #4, (y+k).s == y.s, so k = y.s - (y+k).d On the RHS, (y+k).d can't be computed directly, but y.s can be, and we approximate k by ignoring the (y+k).d term at first. Note that k can't be very large, since all offset-returning methods return a duration of magnitude less than 24 hours. For that reason, if y is firmly in std time, (y+k).d must be 0, so ignoring it has no consequence then. In any case, the new value is z = y + y.s [4] It's helpful to step back at look at [4] from a higher level: it's simply mapping from UTC to tz's standard time. At this point, if z.n - z.o = x.n [5] we have an equivalent time, and are almost done. The insecurity here is at the start of daylight time. Picture US Eastern for concreteness. The wall time jumps from 1:59 to 3:00, and wall hours of the form 2:MM don't make good sense then. The docs ask that an Eastern tzinfo class consider such a time to be EDT (because it's "after 2"), which is a redundant spelling of 1:MM EST on the day DST starts. We want to return the 1:MM EST spelling because that's the only spelling that makes sense on the local wall clock. In fact, if [5] holds at this point, we do have the standard-time spelling, but that takes a bit of proof. We first prove a stronger result. What's the difference between the LHS and RHS of [5]? Let diff = x.n - (z.n - z.o) [6] Now z.n = by [4] (y + y.s).n = by #5 y.n + y.s = since y.n = x.n x.n + y.s = since z and y are have the same tzinfo member, y.s = z.s by #2 x.n + z.s Plugging that back into [6] gives diff = x.n - ((x.n + z.s) - z.o) = expanding x.n - x.n - z.s + z.o = cancelling - z.s + z.o = by #2 z.d So diff = z.d. If [5] is true now, diff = 0, so z.d = 0 too, and we have the standard-time spelling we wanted in the endcase described above. We're done. Contrarily, if z.d = 0, then we have a UTC equivalent, and are also done. If [5] is not true now, diff = z.d != 0, and z.d is the offset we need to add to z (in effect, z is in tz's standard time, and we need to shift the local clock into tz's daylight time). Let z' = z + z.d = z + diff [7] and we can again ask whether z'.n - z'.o = x.n [8] If so, we're done. If not, the tzinfo class is insane, according to the assumptions we've made. This also requires a bit of proof. As before, let's compute the difference between the LHS and RHS of [8] (and skipping some of the justifications for the kinds of substitutions we've done several times already): diff' = x.n - (z'.n - z'.o) = replacing z'.n via [7] x.n - (z.n + diff - z'.o) = replacing diff via [6] x.n - (z.n + x.n - (z.n - z.o) - z'.o) = x.n - z.n - x.n + z.n - z.o + z'.o = cancel x.n - z.n + z.n - z.o + z'.o = cancel z.n - z.o + z'.o = #1 twice -z.s - z.d + z'.s + z'.d = z and z' have same tzinfo z'.d - z.d So z' is UTC-equivalent to x iff z'.d = z.d at this point. If they are equal, we've found the UTC-equivalent so are done. In fact, we stop with [7] and return z', not bothering to compute z'.d. How could z.d and z'd differ? z' = z + z.d [7], so merely moving z' by a dst() offset, and starting from a time already in DST (we know z.d != 0), would have to change the result dst() returns: we start in DST, and moving a little further into it takes us out of DST. There isn't a sane case where this can happen. The closest it gets is at the end of DST, where there's an hour in UTC with no spelling in a hybrid tzinfo class. In US Eastern, that's 5:MM UTC = 0:MM EST = 1:MM EDT. During that hour, on an Eastern clock 1:MM is taken as being in standard time (6:MM UTC) because the docs insist on that, but 0:MM is taken as being in daylight time (4:MM UTC). There is no local time mapping to 5:MM UTC. The local clock jumps from 1:59 back to 1:00 again, and repeats the 1:MM hour in standard time. Since that's what the local clock does, we want to map both UTC hours 5:MM and 6:MM to 1:MM Eastern. The result is ambiguous in local time, but so it goes -- it's the way the local clock works. When x = 5:MM UTC is the input to this algorithm, x.o=0, y.o=-5 and y.d=0, so z=0:MM. z.d=60 (minutes) then, so [5] doesn't hold and we keep going. z' = z + z.d = 1:MM then, and z'.d=0, and z'.d - z.d = -60 != 0 so [8] (correctly) concludes that z' is not UTC-equivalent to x. Because we know z.d said z was in daylight time (else [5] would have held and we would have stopped then), and we know z.d != z'.d (else [8] would have held and we have stopped then), and there are only 2 possible values dst() can return in Eastern, it follows that z'.d must be 0 (which it is in the example, but the reasoning doesn't depend on the example -- it depends on there being two possible dst() outcomes, one zero and the other non-zero). Therefore z' must be in standard time, and is the spelling we want in this case. Note again that z' is not UTC-equivalent as far as the hybrid tzinfo class is concerned (because it takes z' as being in standard time rather than the daylight time we intend here), but returning it gives the real-life "local clock repeats an hour" behavior when mapping the "unspellable" UTC hour into tz. When the input is 6:MM, z=1:MM and z.d=0, and we stop at once, again with the 1:MM standard time spelling we want. So how can this break? One of the assumptions must be violated. Two possibilities: 1) [2] effectively says that y.s is invariant across all y belong to a given time zone. This isn't true if, for political reasons or continental drift, a region decides to change its base offset from UTC. 2) There may be versions of "double daylight" time where the tail end of the analysis gives up a step too early. I haven't thought about that enough to say. In any case, it's clear that the default fromutc() is strong enough to handle "almost all" time zones: so long as the standard offset is invariant, it doesn't matter if daylight time transition points change from year to year, or if daylight time is skipped in some years; it doesn't matter how large or small dst() may get within its bounds; and it doesn't even matter if some perverse time zone returns a negative dst()). So a breaking case must be pretty bizarre, and a tzinfo subclass can override fromutc() if it is. """ try: from _datetime import * except ImportError: pass else: # Clean up unused names del (_DAYNAMES, _DAYS_BEFORE_MONTH, _DAYS_IN_MONTH, _DI100Y, _DI400Y, _DI4Y, _MAXORDINAL, _MONTHNAMES, _build_struct_time, _call_tzinfo_method, _check_date_fields, _check_time_fields, _check_tzinfo_arg, _check_tzname, _check_utc_offset, _cmp, _cmperror, _date_class, _days_before_month, _days_before_year, _days_in_month, _format_time, _is_leap, _isoweek1monday, _math, _ord2ymd, _time, _time_class, _tzinfo_class, _wrap_strftime, _ymd2ord) # XXX Since import * above excludes names that start with _, # docstring does not get overwritten. In the future, it may be # appropriate to maintain a single module level docstring and # remove the following line. from _datetime import __doc__	73.78 KB	2026-01-20 12:58
📁 email	-	2026-03-21 15:45
📁 html	-	2026-03-21 15:45
📁 http	-	2026-03-21 15:45
📄 misc.py """ Miscellaneous function (re)definitions from the Py3.4+ standard library for Python 2.6/2.7. - math.ceil (for Python 2.7) - collections.OrderedDict (for Python 2.6) - collections.Counter (for Python 2.6) - collections.ChainMap (for all versions prior to Python 3.3) - itertools.count (for Python 2.6, with step parameter) - subprocess.check_output (for Python 2.6) - reprlib.recursive_repr (for Python 2.6+) - functools.cmp_to_key (for Python 2.6) """ from __future__ import absolute_import import subprocess from math import ceil as oldceil from operator import itemgetter as _itemgetter, eq as _eq import sys import heapq as _heapq from _weakref import proxy as _proxy from itertools import repeat as _repeat, chain as _chain, starmap as _starmap from socket import getaddrinfo, SOCK_STREAM, error, socket from future.utils import iteritems, itervalues, PY2, PY26, PY3 if PY2: from collections import Mapping, MutableMapping else: from collections.abc import Mapping, MutableMapping def ceil(x): """ Return the ceiling of x as an int. This is the smallest integral value >= x. """ return int(oldceil(x)) ######################################################################## ### reprlib.recursive_repr decorator from Py3.4 ######################################################################## from itertools import islice if PY26: # itertools.count in Py 2.6 doesn't accept a step parameter def count(start=0, step=1): while True: yield start start += step else: from itertools import count if PY3: try: from _thread import get_ident except ImportError: from _dummy_thread import get_ident else: try: from thread import get_ident except ImportError: from dummy_thread import get_ident def recursive_repr(fillvalue='...'): 'Decorator to make a repr function return fillvalue for a recursive call' def decorating_function(user_function): repr_running = set() def wrapper(self): key = id(self), get_ident() if key in repr_running: return fillvalue repr_running.add(key) try: result = user_function(self) finally: repr_running.discard(key) return result # Can't use functools.wraps() here because of bootstrap issues wrapper.__module__ = getattr(user_function, '__module__') wrapper.__doc__ = getattr(user_function, '__doc__') wrapper.__name__ = getattr(user_function, '__name__') wrapper.__annotations__ = getattr(user_function, '__annotations__', {}) return wrapper return decorating_function # OrderedDict Shim from Raymond Hettinger, python core dev # http://code.activestate.com/recipes/576693-ordered-dictionary-for-py24/ # here to support version 2.6. ################################################################################ ### OrderedDict ################################################################################ class _Link(object): __slots__ = 'prev', 'next', 'key', '__weakref__' class OrderedDict(dict): 'Dictionary that remembers insertion order' # An inherited dict maps keys to values. # The inherited dict provides __getitem__, __len__, __contains__, and get. # The remaining methods are order-aware. # Big-O running times for all methods are the same as regular dictionaries. # The internal self.__map dict maps keys to links in a doubly linked list. # The circular doubly linked list starts and ends with a sentinel element. # The sentinel element never gets deleted (this simplifies the algorithm). # The sentinel is in self.__hardroot with a weakref proxy in self.__root. # The prev links are weakref proxies (to prevent circular references). # Individual links are kept alive by the hard reference in self.__map. # Those hard references disappear when a key is deleted from an OrderedDict. def __init__(args, kwds): '''Initialize an ordered dictionary. The signature is the same as regular dictionaries, but keyword arguments are not recommended because their insertion order is arbitrary. ''' if not args: raise TypeError("descriptor '__init__' of 'OrderedDict' object " "needs an argument") self = args[0] args = args[1:] if len(args) > 1: raise TypeError('expected at most 1 arguments, got %d' % len(args)) try: self.__root except AttributeError: self.__hardroot = _Link() self.__root = root = _proxy(self.__hardroot) root.prev = root.next = root self.__map = {} self.__update(args, *kwds) def __setitem__(self, key, value, dict_setitem=dict.__setitem__, proxy=_proxy, Link=_Link): 'od.__setitem__(i, y) <==> od[i]=y' # Setting a new item creates a new link at the end of the linked list, # and the inherited dictionary is updated with the new key/value pair. if key not in self: self.__map[key] = link = Link() root = self.__root last = root.prev link.prev, link.next, link.key = last, root, key last.next = link root.prev = proxy(link) dict_setitem(self, key, value) def __delitem__(self, key, dict_delitem=dict.__delitem__): 'od.__delitem__(y) <==> del od[y]' # Deleting an existing item uses self.__map to find the link which gets # removed by updating the links in the predecessor and successor nodes. dict_delitem(self, key) link = self.__map.pop(key) link_prev = link.prev link_next = link.next link_prev.next = link_next link_next.prev = link_prev def __iter__(self): 'od.__iter__() <==> iter(od)' # Traverse the linked list in order. root = self.__root curr = root.next while curr is not root: yield curr.key curr = curr.next def __reversed__(self): 'od.__reversed__() <==> reversed(od)' # Traverse the linked list in reverse order. root = self.__root curr = root.prev while curr is not root: yield curr.key curr = curr.prev def clear(self): 'od.clear() -> None. Remove all items from od.' root = self.__root root.prev = root.next = root self.__map.clear() dict.clear(self) def popitem(self, last=True): '''od.popitem() -> (k, v), return and remove a (key, value) pair. Pairs are returned in LIFO order if last is true or FIFO order if false. ''' if not self: raise KeyError('dictionary is empty') root = self.__root if last: link = root.prev link_prev = link.prev link_prev.next = root root.prev = link_prev else: link = root.next link_next = link.next root.next = link_next link_next.prev = root key = link.key del self.__map[key] value = dict.pop(self, key) return key, value def move_to_end(self, key, last=True): '''Move an existing element to the end (or beginning if last==False). Raises KeyError if the element does not exist. When last=True, acts like a fast version of self[key]=self.pop(key). ''' link = self.__map[key] link_prev = link.prev link_next = link.next link_prev.next = link_next link_next.prev = link_prev root = self.__root if last: last = root.prev link.prev = last link.next = root last.next = root.prev = link else: first = root.next link.prev = root link.next = first root.next = first.prev = link def __sizeof__(self): sizeof = sys.getsizeof n = len(self) + 1 # number of links including root size = sizeof(self.__dict__) # instance dictionary size += sizeof(self.__map) 2 # internal dict and inherited dict size += sizeof(self.__hardroot) * n # link objects size += sizeof(self.__root) * n # proxy objects return size update = __update = MutableMapping.update keys = MutableMapping.keys values = MutableMapping.values items = MutableMapping.items __ne__ = MutableMapping.__ne__ __marker = object() def pop(self, key, default=__marker): '''od.pop(k[,d]) -> v, remove specified key and return the corresponding value. If key is not found, d is returned if given, otherwise KeyError is raised. ''' if key in self: result = self[key] del self[key] return result if default is self.__marker: raise KeyError(key) return default def setdefault(self, key, default=None): 'od.setdefault(k[,d]) -> od.get(k,d), also set od[k]=d if k not in od' if key in self: return self[key] self[key] = default return default @recursive_repr() def __repr__(self): 'od.__repr__() <==> repr(od)' if not self: return '%s()' % (self.__class__.__name__,) return '%s(%r)' % (self.__class__.__name__, list(self.items())) def __reduce__(self): 'Return state information for pickling' inst_dict = vars(self).copy() for k in vars(OrderedDict()): inst_dict.pop(k, None) return self.__class__, (), inst_dict or None, None, iter(self.items()) def copy(self): 'od.copy() -> a shallow copy of od' return self.__class__(self) @classmethod def fromkeys(cls, iterable, value=None): '''OD.fromkeys(S[, v]) -> New ordered dictionary with keys from S. If not specified, the value defaults to None. ''' self = cls() for key in iterable: self[key] = value return self def __eq__(self, other): '''od.__eq__(y) <==> od==y. Comparison to another OD is order-sensitive while comparison to a regular mapping is order-insensitive. ''' if isinstance(other, OrderedDict): return dict.__eq__(self, other) and all(map(_eq, self, other)) return dict.__eq__(self, other) # {{{ http://code.activestate.com/recipes/576611/ (r11) try: from operator import itemgetter from heapq import nlargest except ImportError: pass ######################################################################## ### Counter ######################################################################## def _count_elements(mapping, iterable): 'Tally elements from the iterable.' mapping_get = mapping.get for elem in iterable: mapping[elem] = mapping_get(elem, 0) + 1 class Counter(dict): '''Dict subclass for counting hashable items. Sometimes called a bag or multiset. Elements are stored as dictionary keys and their counts are stored as dictionary values. >>> c = Counter('abcdeabcdabcaba') # count elements from a string >>> c.most_common(3) # three most common elements [('a', 5), ('b', 4), ('c', 3)] >>> sorted(c) # list all unique elements ['a', 'b', 'c', 'd', 'e'] >>> ''.join(sorted(c.elements())) # list elements with repetitions 'aaaaabbbbcccdde' >>> sum(c.values()) # total of all counts 15 >>> c['a'] # count of letter 'a' 5 >>> for elem in 'shazam': # update counts from an iterable ... c[elem] += 1 # by adding 1 to each element's count >>> c['a'] # now there are seven 'a' 7 >>> del c['b'] # remove all 'b' >>> c['b'] # now there are zero 'b' 0 >>> d = Counter('simsalabim') # make another counter >>> c.update(d) # add in the second counter >>> c['a'] # now there are nine 'a' 9 >>> c.clear() # empty the counter >>> c Counter() Note: If a count is set to zero or reduced to zero, it will remain in the counter until the entry is deleted or the counter is cleared: >>> c = Counter('aaabbc') >>> c['b'] -= 2 # reduce the count of 'b' by two >>> c.most_common() # 'b' is still in, but its count is zero [('a', 3), ('c', 1), ('b', 0)] ''' # References: # http://en.wikipedia.org/wiki/Multiset # http://www.gnu.org/software/smalltalk/manual-base/html_node/Bag.html # http://www.demo2s.com/Tutorial/Cpp/0380__set-multiset/Catalog0380__set-multiset.htm # http://code.activestate.com/recipes/259174/ # Knuth, TAOCP Vol. II section 4.6.3 def __init__(args, kwds): '''Create a new, empty Counter object. And if given, count elements from an input iterable. Or, initialize the count from another mapping of elements to their counts. >>> c = Counter() # a new, empty counter >>> c = Counter('gallahad') # a new counter from an iterable >>> c = Counter({'a': 4, 'b': 2}) # a new counter from a mapping >>> c = Counter(a=4, b=2) # a new counter from keyword args ''' if not args: raise TypeError("descriptor '__init__' of 'Counter' object " "needs an argument") self = args[0] args = args[1:] if len(args) > 1: raise TypeError('expected at most 1 arguments, got %d' % len(args)) super(Counter, self).__init__() self.update(args, kwds) def __missing__(self, key): 'The count of elements not in the Counter is zero.' # Needed so that self[missing_item] does not raise KeyError return 0 def most_common(self, n=None): '''List the n most common elements and their counts from the most common to the least. If n is None, then list all element counts. >>> Counter('abcdeabcdabcaba').most_common(3) [('a', 5), ('b', 4), ('c', 3)] ''' # Emulate Bag.sortedByCount from Smalltalk if n is None: return sorted(self.items(), key=_itemgetter(1), reverse=True) return _heapq.nlargest(n, self.items(), key=_itemgetter(1)) def elements(self): '''Iterator over elements repeating each as many times as its count. >>> c = Counter('ABCABC') >>> sorted(c.elements()) ['A', 'A', 'B', 'B', 'C', 'C'] # Knuth's example for prime factors of 1836: 22 * 3*3 17*1 >>> prime_factors = Counter({2: 2, 3: 3, 17: 1}) >>> product = 1 >>> for factor in prime_factors.elements(): # loop over factors ... product = factor # and multiply them >>> product 1836 Note, if an element's count has been set to zero or is a negative number, elements() will ignore it. ''' # Emulate Bag.do from Smalltalk and Multiset.begin from C++. return _chain.from_iterable(_starmap(_repeat, self.items())) # Override dict methods where necessary @classmethod def fromkeys(cls, iterable, v=None): # There is no equivalent method for counters because setting v=1 # means that no element can have a count greater than one. raise NotImplementedError( 'Counter.fromkeys() is undefined. Use Counter(iterable) instead.') def update(args, kwds): '''Like dict.update() but add counts instead of replacing them. Source can be an iterable, a dictionary, or another Counter instance. >>> c = Counter('which') >>> c.update('witch') # add elements from another iterable >>> d = Counter('watch') >>> c.update(d) # add elements from another counter >>> c['h'] # four 'h' in which, witch, and watch 4 ''' # The regular dict.update() operation makes no sense here because the # replace behavior results in the some of original untouched counts # being mixed-in with all of the other counts for a mismash that # doesn't have a straight-forward interpretation in most counting # contexts. Instead, we implement straight-addition. Both the inputs # and outputs are allowed to contain zero and negative counts. if not args: raise TypeError("descriptor 'update' of 'Counter' object " "needs an argument") self = args[0] args = args[1:] if len(args) > 1: raise TypeError('expected at most 1 arguments, got %d' % len(args)) iterable = args[0] if args else None if iterable is not None: if isinstance(iterable, Mapping): if self: self_get = self.get for elem, count in iterable.items(): self[elem] = count + self_get(elem, 0) else: super(Counter, self).update(iterable) # fast path when counter is empty else: _count_elements(self, iterable) if kwds: self.update(kwds) def subtract(args, *kwds): '''Like dict.update() but subtracts counts instead of replacing them. Counts can be reduced below zero. Both the inputs and outputs are allowed to contain zero and negative counts. Source can be an iterable, a dictionary, or another Counter instance. >>> c = Counter('which') >>> c.subtract('witch') # subtract elements from another iterable >>> c.subtract(Counter('watch')) # subtract elements from another counter >>> c['h'] # 2 in which, minus 1 in witch, minus 1 in watch 0 >>> c['w'] # 1 in which, minus 1 in witch, minus 1 in watch -1 ''' if not args: raise TypeError("descriptor 'subtract' of 'Counter' object " "needs an argument") self = args[0] args = args[1:] if len(args) > 1: raise TypeError('expected at most 1 arguments, got %d' % len(args)) iterable = args[0] if args else None if iterable is not None: self_get = self.get if isinstance(iterable, Mapping): for elem, count in iterable.items(): self[elem] = self_get(elem, 0) - count else: for elem in iterable: self[elem] = self_get(elem, 0) - 1 if kwds: self.subtract(kwds) def copy(self): 'Return a shallow copy.' return self.__class__(self) def __reduce__(self): return self.__class__, (dict(self),) def __delitem__(self, elem): 'Like dict.__delitem__() but does not raise KeyError for missing values.' if elem in self: super(Counter, self).__delitem__(elem) def __repr__(self): if not self: return '%s()' % self.__class__.__name__ try: items = ', '.join(map('%r: %r'.__mod__, self.most_common())) return '%s({%s})' % (self.__class__.__name__, items) except TypeError: # handle case where values are not orderable return '{0}({1!r})'.format(self.__class__.__name__, dict(self)) # Multiset-style mathematical operations discussed in: # Knuth TAOCP Volume II section 4.6.3 exercise 19 # and at http://en.wikipedia.org/wiki/Multiset # # Outputs guaranteed to only include positive counts. # # To strip negative and zero counts, add-in an empty counter: # c += Counter() def __add__(self, other): '''Add counts from two counters. >>> Counter('abbb') + Counter('bcc') Counter({'b': 4, 'c': 2, 'a': 1}) ''' if not isinstance(other, Counter): return NotImplemented result = Counter() for elem, count in self.items(): newcount = count + other[elem] if newcount > 0: result[elem] = newcount for elem, count in other.items(): if elem not in self and count > 0: result[elem] = count return result def __sub__(self, other): ''' Subtract count, but keep only results with positive counts. >>> Counter('abbbc') - Counter('bccd') Counter({'b': 2, 'a': 1}) ''' if not isinstance(other, Counter): return NotImplemented result = Counter() for elem, count in self.items(): newcount = count - other[elem] if newcount > 0: result[elem] = newcount for elem, count in other.items(): if elem not in self and count < 0: result[elem] = 0 - count return result def __or__(self, other): '''Union is the maximum of value in either of the input counters. >>> Counter('abbb') \| Counter('bcc') Counter({'b': 3, 'c': 2, 'a': 1}) ''' if not isinstance(other, Counter): return NotImplemented result = Counter() for elem, count in self.items(): other_count = other[elem] newcount = other_count if count < other_count else count if newcount > 0: result[elem] = newcount for elem, count in other.items(): if elem not in self and count > 0: result[elem] = count return result def __and__(self, other): ''' Intersection is the minimum of corresponding counts. >>> Counter('abbb') & Counter('bcc') Counter({'b': 1}) ''' if not isinstance(other, Counter): return NotImplemented result = Counter() for elem, count in self.items(): other_count = other[elem] newcount = count if count < other_count else other_count if newcount > 0: result[elem] = newcount return result def __pos__(self): 'Adds an empty counter, effectively stripping negative and zero counts' return self + Counter() def __neg__(self): '''Subtracts from an empty counter. Strips positive and zero counts, and flips the sign on negative counts. ''' return Counter() - self def _keep_positive(self): '''Internal method to strip elements with a negative or zero count''' nonpositive = [elem for elem, count in self.items() if not count > 0] for elem in nonpositive: del self[elem] return self def __iadd__(self, other): '''Inplace add from another counter, keeping only positive counts. >>> c = Counter('abbb') >>> c += Counter('bcc') >>> c Counter({'b': 4, 'c': 2, 'a': 1}) ''' for elem, count in other.items(): self[elem] += count return self._keep_positive() def __isub__(self, other): '''Inplace subtract counter, but keep only results with positive counts. >>> c = Counter('abbbc') >>> c -= Counter('bccd') >>> c Counter({'b': 2, 'a': 1}) ''' for elem, count in other.items(): self[elem] -= count return self._keep_positive() def __ior__(self, other): '''Inplace union is the maximum of value from either counter. >>> c = Counter('abbb') >>> c \|= Counter('bcc') >>> c Counter({'b': 3, 'c': 2, 'a': 1}) ''' for elem, other_count in other.items(): count = self[elem] if other_count > count: self[elem] = other_count return self._keep_positive() def __iand__(self, other): '''Inplace intersection is the minimum of corresponding counts. >>> c = Counter('abbb') >>> c &= Counter('bcc') >>> c Counter({'b': 1}) ''' for elem, count in self.items(): other_count = other[elem] if other_count < count: self[elem] = other_count return self._keep_positive() def check_output(popenargs, *kwargs): """ For Python 2.6 compatibility: see http://stackoverflow.com/questions/4814970/ """ if 'stdout' in kwargs: raise ValueError('stdout argument not allowed, it will be overridden.') process = subprocess.Popen(stdout=subprocess.PIPE, popenargs, *kwargs) output, unused_err = process.communicate() retcode = process.poll() if retcode: cmd = kwargs.get("args") if cmd is None: cmd = popenargs[0] raise subprocess.CalledProcessError(retcode, cmd) return output def count(start=0, step=1): """ ``itertools.count`` in Py 2.6 doesn't accept a step parameter. This is an enhanced version of ``itertools.count`` for Py2.6 equivalent to ``itertools.count`` in Python 2.7+. """ while True: yield start start += step ######################################################################## ### ChainMap (helper for configparser and string.Template) ### From the Py3.4 source code. See also: ### https://github.com/kkxue/Py2ChainMap/blob/master/py2chainmap.py ######################################################################## class ChainMap(MutableMapping): ''' A ChainMap groups multiple dicts (or other mappings) together to create a single, updateable view. The underlying mappings are stored in a list. That list is public and can accessed or updated using the maps* attribute. There is no other state. Lookups search the underlying mappings successively until a key is found. In contrast, writes, updates, and deletions only operate on the first mapping. ''' def __init__(self, maps): '''Initialize a ChainMap by setting maps* to the given mappings. If no mappings are provided, a single empty dictionary is used. ''' self.maps = list(maps) or [{}] # always at least one map def __missing__(self, key): raise KeyError(key) def __getitem__(self, key): for mapping in self.maps: try: return mapping[key] # can't use 'key in mapping' with defaultdict except KeyError: pass return self.__missing__(key) # support subclasses that define __missing__ def get(self, key, default=None): return self[key] if key in self else default def __len__(self): return len(set().union(self.maps)) # reuses stored hash values if possible def __iter__(self): return iter(set().union(self.maps)) def __contains__(self, key): return any(key in m for m in self.maps) def __bool__(self): return any(self.maps) # Py2 compatibility: __nonzero__ = __bool__ @recursive_repr() def __repr__(self): return '{0.__class__.__name__}({1})'.format( self, ', '.join(map(repr, self.maps))) @classmethod def fromkeys(cls, iterable, args): 'Create a ChainMap with a single dict created from the iterable.' return cls(dict.fromkeys(iterable, args)) def copy(self): 'New ChainMap or subclass with a new copy of maps[0] and refs to maps[1:]' return self.__class__(self.maps[0].copy(), self.maps[1:]) __copy__ = copy def new_child(self, m=None): # like Django's Context.push() ''' New ChainMap with a new map followed by all previous maps. If no map is provided, an empty dict is used. ''' if m is None: m = {} return self.__class__(m, self.maps) @property def parents(self): # like Django's Context.pop() 'New ChainMap from maps[1:].' return self.__class__(self.maps[1:]) def __setitem__(self, key, value): self.maps[0][key] = value def __delitem__(self, key): try: del self.maps[0][key] except KeyError: raise KeyError('Key not found in the first mapping: {0!r}'.format(key)) def popitem(self): 'Remove and return an item pair from maps[0]. Raise KeyError is maps[0] is empty.' try: return self.maps[0].popitem() except KeyError: raise KeyError('No keys found in the first mapping.') def pop(self, key, args): 'Remove key from maps[0] and return its value. Raise KeyError if key not in maps[0].' try: return self.maps[0].pop(key, args) except KeyError: raise KeyError('Key not found in the first mapping: {0!r}'.format(key)) def clear(self): 'Clear maps[0], leaving maps[1:] intact.' self.maps[0].clear() # Re-use the same sentinel as in the Python stdlib socket module: from socket import _GLOBAL_DEFAULT_TIMEOUT # Was: _GLOBAL_DEFAULT_TIMEOUT = object() def create_connection(address, timeout=_GLOBAL_DEFAULT_TIMEOUT, source_address=None): """Backport of 3-argument create_connection() for Py2.6. Connect to address* and return the socket object. Convenience function. Connect to address (a 2-tuple ``(host, port)``) and return the socket object. Passing the optional timeout parameter will set the timeout on the socket instance before attempting to connect. If no timeout is supplied, the global default timeout setting returned by :func:`getdefaulttimeout` is used. If source_address is set it must be a tuple of (host, port) for the socket to bind as a source address before making the connection. An host of '' or port 0 tells the OS to use the default. """ host, port = address err = None for res in getaddrinfo(host, port, 0, SOCK_STREAM): af, socktype, proto, canonname, sa = res sock = None try: sock = socket(af, socktype, proto) if timeout is not _GLOBAL_DEFAULT_TIMEOUT: sock.settimeout(timeout) if source_address: sock.bind(source_address) sock.connect(sa) return sock except error as _: err = _ if sock is not None: sock.close() if err is not None: raise err else: raise error("getaddrinfo returns an empty list") # Backport from Py2.7 for Py2.6: def cmp_to_key(mycmp): """Convert a cmp= function into a key= function""" class K(object): __slots__ = ['obj'] def __init__(self, obj, *args): self.obj = obj def __lt__(self, other): return mycmp(self.obj, other.obj) < 0 def __gt__(self, other): return mycmp(self.obj, other.obj) > 0 def __eq__(self, other): return mycmp(self.obj, other.obj) == 0 def __le__(self, other): return mycmp(self.obj, other.obj) <= 0 def __ge__(self, other): return mycmp(self.obj, other.obj) >= 0 def __ne__(self, other): return mycmp(self.obj, other.obj) != 0 def __hash__(self): raise TypeError('hash not implemented') return K # Back up our definitions above in case they're useful _OrderedDict = OrderedDict _Counter = Counter _check_output = check_output _count = count _ceil = ceil __count_elements = _count_elements _recursive_repr = recursive_repr _ChainMap = ChainMap _create_connection = create_connection _cmp_to_key = cmp_to_key # Overwrite the definitions above with the usual ones # from the standard library: if sys.version_info >= (2, 7): from collections import OrderedDict, Counter from itertools import count from functools import cmp_to_key try: from subprocess import check_output except ImportError: # Not available. This happens with Google App Engine: see issue #231 pass from socket import create_connection if sys.version_info >= (3, 0): from math import ceil from collections import _count_elements if sys.version_info >= (3, 3): from reprlib import recursive_repr from collections import ChainMap	32.29 KB	2026-01-20 12:58
📄 socket.py # Wrapper module for _socket, providing some additional facilities # implemented in Python. """\ This module provides socket operations and some related functions. On Unix, it supports IP (Internet Protocol) and Unix domain sockets. On other systems, it only supports IP. Functions specific for a socket are available as methods of the socket object. Functions: socket() -- create a new socket object socketpair() -- create a pair of new socket objects [] fromfd() -- create a socket object from an open file descriptor [] fromshare() -- create a socket object from data received from socket.share() [] gethostname() -- return the current hostname gethostbyname() -- map a hostname to its IP number gethostbyaddr() -- map an IP number or hostname to DNS info getservbyname() -- map a service name and a protocol name to a port number getprotobyname() -- map a protocol name (e.g. 'tcp') to a number ntohs(), ntohl() -- convert 16, 32 bit int from network to host byte order htons(), htonl() -- convert 16, 32 bit int from host to network byte order inet_aton() -- convert IP addr string (123.45.67.89) to 32-bit packed format inet_ntoa() -- convert 32-bit packed format IP to string (123.45.67.89) socket.getdefaulttimeout() -- get the default timeout value socket.setdefaulttimeout() -- set the default timeout value create_connection() -- connects to an address, with an optional timeout and optional source address. [] not available on all platforms! Special objects: SocketType -- type object for socket objects error -- exception raised for I/O errors has_ipv6 -- boolean value indicating if IPv6 is supported Integer constants: AF_INET, AF_UNIX -- socket domains (first argument to socket() call) SOCK_STREAM, SOCK_DGRAM, SOCK_RAW -- socket types (second argument) Many other constants may be defined; these may be used in calls to the setsockopt() and getsockopt() methods. """ from __future__ import unicode_literals from __future__ import print_function from __future__ import division from __future__ import absolute_import from future.builtins import super import _socket from _socket import * import os, sys, io try: import errno except ImportError: errno = None EBADF = getattr(errno, 'EBADF', 9) EAGAIN = getattr(errno, 'EAGAIN', 11) EWOULDBLOCK = getattr(errno, 'EWOULDBLOCK', 11) __all__ = ["getfqdn", "create_connection"] __all__.extend(os._get_exports_list(_socket)) _realsocket = socket # WSA error codes if sys.platform.lower().startswith("win"): errorTab = {} errorTab[10004] = "The operation was interrupted." errorTab[10009] = "A bad file handle was passed." errorTab[10013] = "Permission denied." errorTab[10014] = "A fault occurred on the network??" # WSAEFAULT errorTab[10022] = "An invalid operation was attempted." errorTab[10035] = "The socket operation would block" errorTab[10036] = "A blocking operation is already in progress." errorTab[10048] = "The network address is in use." errorTab[10054] = "The connection has been reset." errorTab[10058] = "The network has been shut down." errorTab[10060] = "The operation timed out." errorTab[10061] = "Connection refused." errorTab[10063] = "The name is too long." errorTab[10064] = "The host is down." errorTab[10065] = "The host is unreachable." __all__.append("errorTab") class socket(_socket.socket): """A subclass of _socket.socket adding the makefile() method.""" __slots__ = ["__weakref__", "_io_refs", "_closed"] def __init__(self, family=AF_INET, type=SOCK_STREAM, proto=0, fileno=None): if fileno is None: _socket.socket.__init__(self, family, type, proto) else: _socket.socket.__init__(self, family, type, proto, fileno) self._io_refs = 0 self._closed = False def __enter__(self): return self def __exit__(self, args): if not self._closed: self.close() def __repr__(self): """Wrap __repr__() to reveal the real class name.""" s = _socket.socket.__repr__(self) if s.startswith("<socket object"): s = "<%s.%s%s%s" % (self.__class__.__module__, self.__class__.__name__, getattr(self, '_closed', False) and " [closed] " or "", s[7:]) return s def __getstate__(self): raise TypeError("Cannot serialize socket object") def dup(self): """dup() -> socket object Return a new socket object connected to the same system resource. """ fd = dup(self.fileno()) sock = self.__class__(self.family, self.type, self.proto, fileno=fd) sock.settimeout(self.gettimeout()) return sock def accept(self): """accept() -> (socket object, address info) Wait for an incoming connection. Return a new socket representing the connection, and the address of the client. For IP sockets, the address info is a pair (hostaddr, port). """ fd, addr = self._accept() sock = socket(self.family, self.type, self.proto, fileno=fd) # Issue #7995: if no default timeout is set and the listening # socket had a (non-zero) timeout, force the new socket in blocking # mode to override platform-specific socket flags inheritance. if getdefaulttimeout() is None and self.gettimeout(): sock.setblocking(True) return sock, addr def makefile(self, mode="r", buffering=None, _3to2kwargs): """makefile(...) -> an I/O stream connected to the socket The arguments are as for io.open() after the filename, except the only mode characters supported are 'r', 'w' and 'b'. The semantics are similar too. (XXX refactor to share code?) """ if 'newline' in _3to2kwargs: newline = _3to2kwargs['newline']; del _3to2kwargs['newline'] else: newline = None if 'errors' in _3to2kwargs: errors = _3to2kwargs['errors']; del _3to2kwargs['errors'] else: errors = None if 'encoding' in _3to2kwargs: encoding = _3to2kwargs['encoding']; del _3to2kwargs['encoding'] else: encoding = None for c in mode: if c not in ("r", "w", "b"): raise ValueError("invalid mode %r (only r, w, b allowed)") writing = "w" in mode reading = "r" in mode or not writing assert reading or writing binary = "b" in mode rawmode = "" if reading: rawmode += "r" if writing: rawmode += "w" raw = SocketIO(self, rawmode) self._io_refs += 1 if buffering is None: buffering = -1 if buffering < 0: buffering = io.DEFAULT_BUFFER_SIZE if buffering == 0: if not binary: raise ValueError("unbuffered streams must be binary") return raw if reading and writing: buffer = io.BufferedRWPair(raw, raw, buffering) elif reading: buffer = io.BufferedReader(raw, buffering) else: assert writing buffer = io.BufferedWriter(raw, buffering) if binary: return buffer text = io.TextIOWrapper(buffer, encoding, errors, newline) text.mode = mode return text def _decref_socketios(self): if self._io_refs > 0: self._io_refs -= 1 if self._closed: self.close() def _real_close(self, _ss=_socket.socket): # This function should not reference any globals. See issue #808164. _ss.close(self) def close(self): # This function should not reference any globals. See issue #808164. self._closed = True if self._io_refs <= 0: self._real_close() def detach(self): """detach() -> file descriptor Close the socket object without closing the underlying file descriptor. The object cannot be used after this call, but the file descriptor can be reused for other purposes. The file descriptor is returned. """ self._closed = True return super().detach() def fromfd(fd, family, type, proto=0): """ fromfd(fd, family, type[, proto]) -> socket object Create a socket object from a duplicate of the given file descriptor. The remaining arguments are the same as for socket(). """ nfd = dup(fd) return socket(family, type, proto, nfd) if hasattr(_socket.socket, "share"): def fromshare(info): """ fromshare(info) -> socket object Create a socket object from a the bytes object returned by socket.share(pid). """ return socket(0, 0, 0, info) if hasattr(_socket, "socketpair"): def socketpair(family=None, type=SOCK_STREAM, proto=0): """socketpair([family[, type[, proto]]]) -> (socket object, socket object) Create a pair of socket objects from the sockets returned by the platform socketpair() function. The arguments are the same as for socket() except the default family is AF_UNIX if defined on the platform; otherwise, the default is AF_INET. """ if family is None: try: family = AF_UNIX except NameError: family = AF_INET a, b = _socket.socketpair(family, type, proto) a = socket(family, type, proto, a.detach()) b = socket(family, type, proto, b.detach()) return a, b _blocking_errnos = set([EAGAIN, EWOULDBLOCK]) class SocketIO(io.RawIOBase): """Raw I/O implementation for stream sockets. This class supports the makefile() method on sockets. It provides the raw I/O interface on top of a socket object. """ # One might wonder why not let FileIO do the job instead. There are two # main reasons why FileIO is not adapted: # - it wouldn't work under Windows (where you can't used read() and # write() on a socket handle) # - it wouldn't work with socket timeouts (FileIO would ignore the # timeout and consider the socket non-blocking) # XXX More docs def __init__(self, sock, mode): if mode not in ("r", "w", "rw", "rb", "wb", "rwb"): raise ValueError("invalid mode: %r" % mode) io.RawIOBase.__init__(self) self._sock = sock if "b" not in mode: mode += "b" self._mode = mode self._reading = "r" in mode self._writing = "w" in mode self._timeout_occurred = False def readinto(self, b): """Read up to len(b) bytes into the writable buffer b* and return the number of bytes read. If the socket is non-blocking and no bytes are available, None is returned. If b is non-empty, a 0 return value indicates that the connection was shutdown at the other end. """ self._checkClosed() self._checkReadable() if self._timeout_occurred: raise IOError("cannot read from timed out object") while True: try: return self._sock.recv_into(b) except timeout: self._timeout_occurred = True raise # except InterruptedError: # continue except error as e: if e.args[0] in _blocking_errnos: return None raise def write(self, b): """Write the given bytes or bytearray object b to the socket and return the number of bytes written. This can be less than len(b) if not all data could be written. If the socket is non-blocking and no bytes could be written None is returned. """ self._checkClosed() self._checkWritable() try: return self._sock.send(b) except error as e: # XXX what about EINTR? if e.args[0] in _blocking_errnos: return None raise def readable(self): """True if the SocketIO is open for reading. """ if self.closed: raise ValueError("I/O operation on closed socket.") return self._reading def writable(self): """True if the SocketIO is open for writing. """ if self.closed: raise ValueError("I/O operation on closed socket.") return self._writing def seekable(self): """True if the SocketIO is open for seeking. """ if self.closed: raise ValueError("I/O operation on closed socket.") return super().seekable() def fileno(self): """Return the file descriptor of the underlying socket. """ self._checkClosed() return self._sock.fileno() @property def name(self): if not self.closed: return self.fileno() else: return -1 @property def mode(self): return self._mode def close(self): """Close the SocketIO object. This doesn't close the underlying socket, except if all references to it have disappeared. """ if self.closed: return io.RawIOBase.close(self) self._sock._decref_socketios() self._sock = None def getfqdn(name=''): """Get fully qualified domain name from name. An empty argument is interpreted as meaning the local host. First the hostname returned by gethostbyaddr() is checked, then possibly existing aliases. In case no FQDN is available, hostname from gethostname() is returned. """ name = name.strip() if not name or name == '0.0.0.0': name = gethostname() try: hostname, aliases, ipaddrs = gethostbyaddr(name) except error: pass else: aliases.insert(0, hostname) for name in aliases: if '.' in name: break else: name = hostname return name # Re-use the same sentinel as in the Python stdlib socket module: from socket import _GLOBAL_DEFAULT_TIMEOUT # Was: _GLOBAL_DEFAULT_TIMEOUT = object() def create_connection(address, timeout=_GLOBAL_DEFAULT_TIMEOUT, source_address=None): """Connect to address and return the socket object. Convenience function. Connect to address (a 2-tuple ``(host, port)``) and return the socket object. Passing the optional timeout parameter will set the timeout on the socket instance before attempting to connect. If no timeout is supplied, the global default timeout setting returned by :func:`getdefaulttimeout` is used. If source_address is set it must be a tuple of (host, port) for the socket to bind as a source address before making the connection. An host of '' or port 0 tells the OS to use the default. """ host, port = address err = None for res in getaddrinfo(host, port, 0, SOCK_STREAM): af, socktype, proto, canonname, sa = res sock = None try: sock = socket(af, socktype, proto) if timeout is not _GLOBAL_DEFAULT_TIMEOUT: sock.settimeout(timeout) if source_address: sock.bind(source_address) sock.connect(sa) return sock except error as _: err = _ if sock is not None: sock.close() if err is not None: raise err else: raise error("getaddrinfo returns an empty list")	15.3 KB	2026-01-20 12:58
📄 socketserver.py """Generic socket server classes. This module tries to capture the various aspects of defining a server: For socket-based servers: - address family: - AF_INET{,6}: IP (Internet Protocol) sockets (default) - AF_UNIX: Unix domain sockets - others, e.g. AF_DECNET are conceivable (see <socket.h> - socket type: - SOCK_STREAM (reliable stream, e.g. TCP) - SOCK_DGRAM (datagrams, e.g. UDP) For request-based servers (including socket-based): - client address verification before further looking at the request (This is actually a hook for any processing that needs to look at the request before anything else, e.g. logging) - how to handle multiple requests: - synchronous (one request is handled at a time) - forking (each request is handled by a new process) - threading (each request is handled by a new thread) The classes in this module favor the server type that is simplest to write: a synchronous TCP/IP server. This is bad class design, but save some typing. (There's also the issue that a deep class hierarchy slows down method lookups.) There are five classes in an inheritance diagram, four of which represent synchronous servers of four types: +------------+ \| BaseServer \| +------------+ \| v +-----------+ +------------------+ \| TCPServer \|------->\| UnixStreamServer \| +-----------+ +------------------+ \| v +-----------+ +--------------------+ \| UDPServer \|------->\| UnixDatagramServer \| +-----------+ +--------------------+ Note that UnixDatagramServer derives from UDPServer, not from UnixStreamServer -- the only difference between an IP and a Unix stream server is the address family, which is simply repeated in both unix server classes. Forking and threading versions of each type of server can be created using the ForkingMixIn and ThreadingMixIn mix-in classes. For instance, a threading UDP server class is created as follows: class ThreadingUDPServer(ThreadingMixIn, UDPServer): pass The Mix-in class must come first, since it overrides a method defined in UDPServer! Setting the various member variables also changes the behavior of the underlying server mechanism. To implement a service, you must derive a class from BaseRequestHandler and redefine its handle() method. You can then run various versions of the service by combining one of the server classes with your request handler class. The request handler class must be different for datagram or stream services. This can be hidden by using the request handler subclasses StreamRequestHandler or DatagramRequestHandler. Of course, you still have to use your head! For instance, it makes no sense to use a forking server if the service contains state in memory that can be modified by requests (since the modifications in the child process would never reach the initial state kept in the parent process and passed to each child). In this case, you can use a threading server, but you will probably have to use locks to avoid two requests that come in nearly simultaneous to apply conflicting changes to the server state. On the other hand, if you are building e.g. an HTTP server, where all data is stored externally (e.g. in the file system), a synchronous class will essentially render the service "deaf" while one request is being handled -- which may be for a very long time if a client is slow to read all the data it has requested. Here a threading or forking server is appropriate. In some cases, it may be appropriate to process part of a request synchronously, but to finish processing in a forked child depending on the request data. This can be implemented by using a synchronous server and doing an explicit fork in the request handler class handle() method. Another approach to handling multiple simultaneous requests in an environment that supports neither threads nor fork (or where these are too expensive or inappropriate for the service) is to maintain an explicit table of partially finished requests and to use select() to decide which request to work on next (or whether to handle a new incoming request). This is particularly important for stream services where each client can potentially be connected for a long time (if threads or subprocesses cannot be used). Future work: - Standard classes for Sun RPC (which uses either UDP or TCP) - Standard mix-in classes to implement various authentication and encryption schemes - Standard framework for select-based multiplexing XXX Open problems: - What to do with out-of-band data? BaseServer: - split generic "request" functionality out into BaseServer class. Copyright (C) 2000 Luke Kenneth Casson Leighton <lkcl@samba.org> example: read entries from a SQL database (requires overriding get_request() to return a table entry from the database). entry is processed by a RequestHandlerClass. """ # Author of the BaseServer patch: Luke Kenneth Casson Leighton # XXX Warning! # There is a test suite for this module, but it cannot be run by the # standard regression test. # To run it manually, run Lib/test/test_socketserver.py. from __future__ import (absolute_import, print_function) __version__ = "0.4" import socket import select import sys import os import errno try: import threading except ImportError: import dummy_threading as threading __all__ = ["TCPServer","UDPServer","ForkingUDPServer","ForkingTCPServer", "ThreadingUDPServer","ThreadingTCPServer","BaseRequestHandler", "StreamRequestHandler","DatagramRequestHandler", "ThreadingMixIn", "ForkingMixIn"] if hasattr(socket, "AF_UNIX"): __all__.extend(["UnixStreamServer","UnixDatagramServer", "ThreadingUnixStreamServer", "ThreadingUnixDatagramServer"]) def _eintr_retry(func, args): """restart a system call interrupted by EINTR""" while True: try: return func(args) except OSError as e: if e.errno != errno.EINTR: raise class BaseServer(object): """Base class for server classes. Methods for the caller: - __init__(server_address, RequestHandlerClass) - serve_forever(poll_interval=0.5) - shutdown() - handle_request() # if you do not use serve_forever() - fileno() -> int # for select() Methods that may be overridden: - server_bind() - server_activate() - get_request() -> request, client_address - handle_timeout() - verify_request(request, client_address) - server_close() - process_request(request, client_address) - shutdown_request(request) - close_request(request) - service_actions() - handle_error() Methods for derived classes: - finish_request(request, client_address) Class variables that may be overridden by derived classes or instances: - timeout - address_family - socket_type - allow_reuse_address Instance variables: - RequestHandlerClass - socket """ timeout = None def __init__(self, server_address, RequestHandlerClass): """Constructor. May be extended, do not override.""" self.server_address = server_address self.RequestHandlerClass = RequestHandlerClass self.__is_shut_down = threading.Event() self.__shutdown_request = False def server_activate(self): """Called by constructor to activate the server. May be overridden. """ pass def serve_forever(self, poll_interval=0.5): """Handle one request at a time until shutdown. Polls for shutdown every poll_interval seconds. Ignores self.timeout. If you need to do periodic tasks, do them in another thread. """ self.__is_shut_down.clear() try: while not self.__shutdown_request: # XXX: Consider using another file descriptor or # connecting to the socket to wake this up instead of # polling. Polling reduces our responsiveness to a # shutdown request and wastes cpu at all other times. r, w, e = _eintr_retry(select.select, [self], [], [], poll_interval) if self in r: self._handle_request_noblock() self.service_actions() finally: self.__shutdown_request = False self.__is_shut_down.set() def shutdown(self): """Stops the serve_forever loop. Blocks until the loop has finished. This must be called while serve_forever() is running in another thread, or it will deadlock. """ self.__shutdown_request = True self.__is_shut_down.wait() def service_actions(self): """Called by the serve_forever() loop. May be overridden by a subclass / Mixin to implement any code that needs to be run during the loop. """ pass # The distinction between handling, getting, processing and # finishing a request is fairly arbitrary. Remember: # # - handle_request() is the top-level call. It calls # select, get_request(), verify_request() and process_request() # - get_request() is different for stream or datagram sockets # - process_request() is the place that may fork a new process # or create a new thread to finish the request # - finish_request() instantiates the request handler class; # this constructor will handle the request all by itself def handle_request(self): """Handle one request, possibly blocking. Respects self.timeout. """ # Support people who used socket.settimeout() to escape # handle_request before self.timeout was available. timeout = self.socket.gettimeout() if timeout is None: timeout = self.timeout elif self.timeout is not None: timeout = min(timeout, self.timeout) fd_sets = _eintr_retry(select.select, [self], [], [], timeout) if not fd_sets[0]: self.handle_timeout() return self._handle_request_noblock() def _handle_request_noblock(self): """Handle one request, without blocking. I assume that select.select has returned that the socket is readable before this function was called, so there should be no risk of blocking in get_request(). """ try: request, client_address = self.get_request() except socket.error: return if self.verify_request(request, client_address): try: self.process_request(request, client_address) except: self.handle_error(request, client_address) self.shutdown_request(request) def handle_timeout(self): """Called if no new request arrives within self.timeout. Overridden by ForkingMixIn. """ pass def verify_request(self, request, client_address): """Verify the request. May be overridden. Return True if we should proceed with this request. """ return True def process_request(self, request, client_address): """Call finish_request. Overridden by ForkingMixIn and ThreadingMixIn. """ self.finish_request(request, client_address) self.shutdown_request(request) def server_close(self): """Called to clean-up the server. May be overridden. """ pass def finish_request(self, request, client_address): """Finish one request by instantiating RequestHandlerClass.""" self.RequestHandlerClass(request, client_address, self) def shutdown_request(self, request): """Called to shutdown and close an individual request.""" self.close_request(request) def close_request(self, request): """Called to clean up an individual request.""" pass def handle_error(self, request, client_address): """Handle an error gracefully. May be overridden. The default is to print a traceback and continue. """ print('-'40) print('Exception happened during processing of request from', end=' ') print(client_address) import traceback traceback.print_exc() # XXX But this goes to stderr! print('-'40) class TCPServer(BaseServer): """Base class for various socket-based server classes. Defaults to synchronous IP stream (i.e., TCP). Methods for the caller: - __init__(server_address, RequestHandlerClass, bind_and_activate=True) - serve_forever(poll_interval=0.5) - shutdown() - handle_request() # if you don't use serve_forever() - fileno() -> int # for select() Methods that may be overridden: - server_bind() - server_activate() - get_request() -> request, client_address - handle_timeout() - verify_request(request, client_address) - process_request(request, client_address) - shutdown_request(request) - close_request(request) - handle_error() Methods for derived classes: - finish_request(request, client_address) Class variables that may be overridden by derived classes or instances: - timeout - address_family - socket_type - request_queue_size (only for stream sockets) - allow_reuse_address Instance variables: - server_address - RequestHandlerClass - socket """ address_family = socket.AF_INET socket_type = socket.SOCK_STREAM request_queue_size = 5 allow_reuse_address = False def __init__(self, server_address, RequestHandlerClass, bind_and_activate=True): """Constructor. May be extended, do not override.""" BaseServer.__init__(self, server_address, RequestHandlerClass) self.socket = socket.socket(self.address_family, self.socket_type) if bind_and_activate: self.server_bind() self.server_activate() def server_bind(self): """Called by constructor to bind the socket. May be overridden. """ if self.allow_reuse_address: self.socket.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1) self.socket.bind(self.server_address) self.server_address = self.socket.getsockname() def server_activate(self): """Called by constructor to activate the server. May be overridden. """ self.socket.listen(self.request_queue_size) def server_close(self): """Called to clean-up the server. May be overridden. """ self.socket.close() def fileno(self): """Return socket file number. Interface required by select(). """ return self.socket.fileno() def get_request(self): """Get the request and client address from the socket. May be overridden. """ return self.socket.accept() def shutdown_request(self, request): """Called to shutdown and close an individual request.""" try: #explicitly shutdown. socket.close() merely releases #the socket and waits for GC to perform the actual close. request.shutdown(socket.SHUT_WR) except socket.error: pass #some platforms may raise ENOTCONN here self.close_request(request) def close_request(self, request): """Called to clean up an individual request.""" request.close() class UDPServer(TCPServer): """UDP server class.""" allow_reuse_address = False socket_type = socket.SOCK_DGRAM max_packet_size = 8192 def get_request(self): data, client_addr = self.socket.recvfrom(self.max_packet_size) return (data, self.socket), client_addr def server_activate(self): # No need to call listen() for UDP. pass def shutdown_request(self, request): # No need to shutdown anything. self.close_request(request) def close_request(self, request): # No need to close anything. pass class ForkingMixIn(object): """Mix-in class to handle each request in a new process.""" timeout = 300 active_children = None max_children = 40 def collect_children(self): """Internal routine to wait for children that have exited.""" if self.active_children is None: return while len(self.active_children) >= self.max_children: # XXX: This will wait for any child process, not just ones # spawned by this library. This could confuse other # libraries that expect to be able to wait for their own # children. try: pid, status = os.waitpid(0, 0) except os.error: pid = None if pid not in self.active_children: continue self.active_children.remove(pid) # XXX: This loop runs more system calls than it ought # to. There should be a way to put the active_children into a # process group and then use os.waitpid(-pgid) to wait for any # of that set, but I couldn't find a way to allocate pgids # that couldn't collide. for child in self.active_children: try: pid, status = os.waitpid(child, os.WNOHANG) except os.error: pid = None if not pid: continue try: self.active_children.remove(pid) except ValueError as e: raise ValueError('%s. x=%d and list=%r' % (e.message, pid, self.active_children)) def handle_timeout(self): """Wait for zombies after self.timeout seconds of inactivity. May be extended, do not override. """ self.collect_children() def service_actions(self): """Collect the zombie child processes regularly in the ForkingMixIn. service_actions is called in the BaseServer's serve_forver loop. """ self.collect_children() def process_request(self, request, client_address): """Fork a new subprocess to process the request.""" pid = os.fork() if pid: # Parent process if self.active_children is None: self.active_children = [] self.active_children.append(pid) self.close_request(request) return else: # Child process. # This must never return, hence os._exit()! try: self.finish_request(request, client_address) self.shutdown_request(request) os._exit(0) except: try: self.handle_error(request, client_address) self.shutdown_request(request) finally: os._exit(1) class ThreadingMixIn(object): """Mix-in class to handle each request in a new thread.""" # Decides how threads will act upon termination of the # main process daemon_threads = False def process_request_thread(self, request, client_address): """Same as in BaseServer but as a thread. In addition, exception handling is done here. """ try: self.finish_request(request, client_address) self.shutdown_request(request) except: self.handle_error(request, client_address) self.shutdown_request(request) def process_request(self, request, client_address): """Start a new thread to process the request.""" t = threading.Thread(target = self.process_request_thread, args = (request, client_address)) t.daemon = self.daemon_threads t.start() class ForkingUDPServer(ForkingMixIn, UDPServer): pass class ForkingTCPServer(ForkingMixIn, TCPServer): pass class ThreadingUDPServer(ThreadingMixIn, UDPServer): pass class ThreadingTCPServer(ThreadingMixIn, TCPServer): pass if hasattr(socket, 'AF_UNIX'): class UnixStreamServer(TCPServer): address_family = socket.AF_UNIX class UnixDatagramServer(UDPServer): address_family = socket.AF_UNIX class ThreadingUnixStreamServer(ThreadingMixIn, UnixStreamServer): pass class ThreadingUnixDatagramServer(ThreadingMixIn, UnixDatagramServer): pass class BaseRequestHandler(object): """Base class for request handler classes. This class is instantiated for each request to be handled. The constructor sets the instance variables request, client_address and server, and then calls the handle() method. To implement a specific service, all you need to do is to derive a class which defines a handle() method. The handle() method can find the request as self.request, the client address as self.client_address, and the server (in case it needs access to per-server information) as self.server. Since a separate instance is created for each request, the handle() method can define arbitrary other instance variariables. """ def __init__(self, request, client_address, server): self.request = request self.client_address = client_address self.server = server self.setup() try: self.handle() finally: self.finish() def setup(self): pass def handle(self): pass def finish(self): pass # The following two classes make it possible to use the same service # class for stream or datagram servers. # Each class sets up these instance variables: # - rfile: a file object from which receives the request is read # - wfile: a file object to which the reply is written # When the handle() method returns, wfile is flushed properly class StreamRequestHandler(BaseRequestHandler): """Define self.rfile and self.wfile for stream sockets.""" # Default buffer sizes for rfile, wfile. # We default rfile to buffered because otherwise it could be # really slow for large data (a getc() call per byte); we make # wfile unbuffered because (a) often after a write() we want to # read and we need to flush the line; (b) big writes to unbuffered # files are typically optimized by stdio even when big reads # aren't. rbufsize = -1 wbufsize = 0 # A timeout to apply to the request socket, if not None. timeout = None # Disable nagle algorithm for this socket, if True. # Use only when wbufsize != 0, to avoid small packets. disable_nagle_algorithm = False def setup(self): self.connection = self.request if self.timeout is not None: self.connection.settimeout(self.timeout) if self.disable_nagle_algorithm: self.connection.setsockopt(socket.IPPROTO_TCP, socket.TCP_NODELAY, True) self.rfile = self.connection.makefile('rb', self.rbufsize) self.wfile = self.connection.makefile('wb', self.wbufsize) def finish(self): if not self.wfile.closed: try: self.wfile.flush() except socket.error: # An final socket error may have occurred here, such as # the local error ECONNABORTED. pass self.wfile.close() self.rfile.close() class DatagramRequestHandler(BaseRequestHandler): # XXX Regrettably, I cannot get this working on Linux; # s.recvfrom() doesn't return a meaningful client address. """Define self.rfile and self.wfile for datagram sockets.""" def setup(self): from io import BytesIO self.packet, self.socket = self.request self.rfile = BytesIO(self.packet) self.wfile = BytesIO() def finish(self): self.socket.sendto(self.wfile.getvalue(), self.client_address)	23.72 KB	2026-01-20 12:58
📁 test	-	2026-03-21 15:45
📄 total_ordering.py """ For Python < 2.7.2. total_ordering in versions prior to 2.7.2 is buggy. See http://bugs.python.org/issue10042 for details. For these versions use code borrowed from Python 2.7.3. From django.utils. """ import sys if sys.version_info >= (2, 7, 2): from functools import total_ordering else: def total_ordering(cls): """Class decorator that fills in missing ordering methods""" convert = { '__lt__': [('__gt__', lambda self, other: not (self < other or self == other)), ('__le__', lambda self, other: self < other or self == other), ('__ge__', lambda self, other: not self < other)], '__le__': [('__ge__', lambda self, other: not self <= other or self == other), ('__lt__', lambda self, other: self <= other and not self == other), ('__gt__', lambda self, other: not self <= other)], '__gt__': [('__lt__', lambda self, other: not (self > other or self == other)), ('__ge__', lambda self, other: self > other or self == other), ('__le__', lambda self, other: not self > other)], '__ge__': [('__le__', lambda self, other: (not self >= other) or self == other), ('__gt__', lambda self, other: self >= other and not self == other), ('__lt__', lambda self, other: not self >= other)] } roots = set(dir(cls)) & set(convert) if not roots: raise ValueError('must define at least one ordering operation: < > <= >=') root = max(roots) # prefer __lt__ to __le__ to __gt__ to __ge__ for opname, opfunc in convert[root]: if opname not in roots: opfunc.__name__ = opname opfunc.__doc__ = getattr(int, opname).__doc__ setattr(cls, opname, opfunc) return cls	1.88 KB	2026-01-20 12:58
📁 urllib	-	2026-03-21 15:45
📁 xmlrpc	-	2026-03-21 15:45

Terminal