Python Cookbook

# unpacking
data = ["ACME", 50, 91.1, (2012, 12, 21)]
name, *_, (year, mon, day) = data

# fixed-length queue
from collections import deque

d = deque(maxlen=N)

# heap related
import heapq

heapq.nlargest(N, items)  # faster than sorted(items)[:N] if N is small
heapq.nsmallest(N, items, key=None)
nums = [1, 8, 2, 23, 7, -4, 18, 23, 42, 37, 2]
heapq.heapify(nums)  # inplace function

# agg by values
prices = {"AAPL": 612.78, "IBM": 205.55, "FB": 10.75}
min_price = min(zip(prices.values(), prices.keys()))  # zip to ((value, key)) generator

# key-views support set oparations
d.keys() - d2.keys()
d.keys() & d2.keys()
d.keys() | d2.keys()

# boolean selectors
from itertools import compress
alist = list(range(10))
list(compress(alist, [i % 2 == 0 for i in alist]))

import heapq

class PriorityQueue:
    def __init__(self):
	self._queue = []
	self._index = 0 # using index to properly order items with the same priority level

    def push(self, item, priority):
	heapq.heappush(self._queue, (-priority, self._index, item))
	self._index += 1

    def pop(self):
	return heapq.heappop(self._queue)[-1]

import re

re.split(r"[./]", "btc.usd/coinbase")
# => ['btc', 'usd', 'coinbase'] slower than split("/").split(".")

# shell-like matching
fnmatch("foo.txt", "*.txt")

# searching & replacing
# str.find, str.findall, re.match
# str.replace, re.sub
# Case-Insensitive Option: flags=re.IGNORECASE


# For the Shortest Match
str_pat = re.compile(r"\"(.*)\"")  # default: greedy regular expression
text = 'Computer says "no." Phone says "yes."'
str_pat.findall(text)
# => ['no." Phone says "yes.']

str_pat = re.compile(r"\"(.*?)\"")  # nongreedy and produces the shortest match instead.
str_pat.findall(text)
# => ['no.', 'yes.']

# Aligning
# str.ljust, str.rjust, str.center
# f"{text:<20}", f"{text:>20}", f"{text:^20}"
text = "Hello World"
ftext = f"{text:=>20}"  # => "=========Hello World"
ftext = f"{text:*^20}"  # => "****Hello World*****"

# Byte Strings
bstr = b"Hello World"
barray = bytearray(b"Hello World")
bstr.decode("ascii") # => "Hello World"

text = "foo = 23 + 42 * 10"

# expected results
tokens = [
    ("NAME", "foo"),
    ("EQ", "="),
    ("NUM", "23"),
    ("PLUS", "+"),
    ("NUM", "42"),
    ("TIMES", "*"),
    ("NUM", "10"),
]

import re

NAME = r"(?P<NAME>[a-zA-Z_][a-zA-Z_0-9]*)"
NUM = r"(?P<NUM>\d+)"
PLUS = r"(?P<PLUS>\+)"
TIMES = r"(?P<TIMES>\*)"
EQ = r"(?P<EQ>=)"
WS = r"(?P<WS>\s+)"
# the ?P<TOKENNAME> convention is used to assign a name to the pattern
master_pat = re.compile("|".join([NAME, NUM, PLUS, TIMES, EQ, WS]))

for mo in re.finditer(master_pat, "foo = 42"):
    print(mo.lastgroup, mo.group())  # NAME foo, WS , EQ =, ...

round(1.29, 1)
# => 1.3
round(1245, -1)
# => 1240
round(1275, -1)
# => 1280

from decimal import Decimal, localcontext
a = Decimal('6.32')
b = Decimal('2.41')

with localcontext() as ctx:
    ctx.prec = 5
    print(a/b) # 2.6224

x = 1234.56789
format(x, '0.2f')
# => '1234.57'   # round
format(x, '>10.1f')
# => '    1234.6'
format(x, '0,.1f')
# => '1,234.6

x = 1234
bin(x)  # others: oct, hex
# => '0b10011010010'
format(x, 'b')  # others: o, x
# => '0011010010'
int('10011010010', 2)
# => 1234

x = -1234
format(x, 'b')
#=> '-10011010010'
format(x, 'x')
#=> '-4d2'
format(2**32 + x, 'b')
#=> '11111111111111111111101100101110'
format(2**32 + x, 'x')
#=> 'fffffb2e'
int('4d2', 16)
#=> 1234
int('10011010010', 2)
#=> 1234

data = b'\x00\x124V\x00x\x90\xab\x00\xcd\xef\x01\x00#\x004'
x = int.from_bytes(data, 'little')  # or 'big

x = 94522842520747284487117727783387188
x.to_bytes(16, 'little')

a = complex(2, 4)
b = 3 - 5j
a.conjugate()
#=> (2-4j)
abs(a)
#=> 4.47213595499958
a * b
#=> (26+2j)

import cmath
cmath.sin(a)
#=> (24.83130584894638-11.356612711218174j)

import numpy as np
a = np.array([2 + 3j, 4 + 5j, 6 - 7j, 8 + 9j])
np.sin(a)

from dateutil.relativedelta import relativedelta
from dateutil.rrule import FR
d = datetime.now()
print(d + relativedelta(weekday=FR(-1)))

import pytz
d = datetime.now() # no timezone info
print(d)
# => 2018-12-21 17:14:01.258941

shanghai = pytz.timezone('Asia/Shanghai')
loc_d = shanghai.localize(d) # Localize the date for Shanghai
print(loc_d)
# => 2018-12-21 17:14:01.258941+08:00

# Once the date has been localized, it can be converted to other time zones
utc_d = loc_d.astimezone(pytz.utc)
print(utc_d)
# => 2018-12-21 09:14:01.258941+00:00

iterable = iter(range(5))  # Invokes range.__iter__()
try:
    while True:
	line = next(iterable)  # Invokes iterable.__next__()
	print(line, end='')
except StopIteration:
    pass

# non exception version
while True:
    line = next(iterable, None)
    if line is None:
	break
    print(line, end='')

a = [1, 2, 3]
b = ['w', 'x', 'y', 'z']

for i in zip(a, b):
    print(i)
#=> (1, 'w') (2, 'x') (3, 'y')

from itertools import zip_longest
for i in zip_longest(a, b):
    print(i)
#=> (1, 'w') (2, 'x') (3, 'y') (None, 'z')

from collections import Iterable


def flatten(items, ignore_types=(str, bytes)):
    for x in items:
	if isinstance(x, Iterable) and not isinstance(x, ignore_types):
	    yield from flatten(x)
	else:
	    yield x

items = ['Dave', 'Paula', ['Thomas', 'Lewis']]
for x in flatten(items):
    print(x)

import heapq
a = [1, 4, 7, 10]
b = [2, 5, 6, 11]
for c in heapq.merge(a, b):
    print(c)

for chunk in iter(lambda: fs.read(10), ''):
    print(chunk)

with open('somefile.txt', 'rt', encoding='latin-1') as f:
    ...

# Replace bad chars with Unicode U+fffd replacement char
open('sample.txt', 'rt', encoding='ascii', errors='replace')

# Ignore bad chars entirely
open('sample.txt', 'rt', encoding='ascii', errors='ignore')

import array
a = array.array('i', [0, 0, 0, 0, 0, 0, 0, 0])
with open('data.bin', 'rb') as f:
    f.readinto(a)

from functools import partial

RECORD_SIZE = 32

with open('somefile.data', 'rb') as f:
    records = iter(partial(f.read, RECORD_SIZE), b'')
    for r in records:
	...

os.path.basename(path)
os.path.dirname(path)
os.path.expanduser(path)
os.path.splitext(path)  # Split the file extension
os.path.exists(path)
os.path.isfile(path) # isdir, islink
os.path.realpath('/usr/local/bin/python3') # => '/usr/local/bin/python3.3'
os.path.getsize() # getmtime
os.listdir(dir)

import io
# decode a binary file
with open('some_binary_file.bin', 'rb') as open_file:
    fs = io.TextIOWrapper(open_file, encoding='utf8')
    text = fs.read()

# change encoding
import sys
sys.stdout.encoding  #=> 'UTF-8'
# use its detach() method to remove the existing text encoding layer before replacing it with a new one
sys.stdout = io.TextIOWrapper(sys.stdout.detach(), encoding='latin-1')
sys.stdout.encoding #=> 'latin-1'

f = open('sample.txt', 'w')
# a text-handling layer that encodes and decodes Unicode
f # => <_io.TextIOWrapper name='sample.txt' mode='w' encoding='UTF-8'>

# a buffered I/O layer that handles binary data
f.buffer # => <_io.BufferedWriter name='sample.txt'>
f.buffer.write(b'hello\n') # write bytes to a text file

# io.FileIO is a raw file representing the low-level file descriptor in the operating system
f.buffer.raw # => <_io.FileIO name='sample.txt' mode='wb'>

# Create a file object, but don't close underlying fd when done
f = open(fd, 'wt', closefd=False)

def echo_client(client_sock, addr):
    print('Got connection from', addr)

    # Make text-mode file wrappers for socket reading/writing, only works on Unix-based systems
    # Use the makefile() method of sockets instead to be cross platform
    client_in = open(client_sock.fileno(), 'rt', encoding='latin-1',
			 closefd=False)
    client_out = open(client_sock.fileno(), 'wt', encoding='latin-1',
			  closefd=False)

    # Echo lines back to the client using file I/O
    for line in client_in:
	client_out.write(line)
	client_out.flush()
    client_sock.close()

from tempfile import TemporaryFile, NamedTemporaryFile, TemporaryDirectory
with TemporaryFile('w+t', encoding='utf-8', errors='ignore') as f:
    f.write('Hello World\n')

with NamedTemporaryFile(
	'w+t', delete=False, prefix='mytemp', suffix='.txt', dir='/tmp') as f:
    print('filename is:', f.name)  #=> /tmp/mytemp2tmz4nl5.txt

with TemporaryDirectory() as dirname:
    print('dirname is:', dirname)

import json
from collections import OrderedDict
s = '{"name": "ACME", "shares": 50, "price": 490.1}'
data = json.loads(s, object_pairs_hook=OrderedDict)
# data => OrderedDict([('name', 'ACME'), ('shares', 50), ('price', 490.1)])

class JSONObject:
    def __init__(self, d):
	self.__dict__ = d

    def __str__(self):
	return str(self.__dict__)


obj = json.loads(s, object_hook=JSONObject)
obj.name = 'def'

json.dumps(vars(obj)) # vars(obj) same as obj.__dict__
# or
json.dumps(obj, default=vars) # use vars as a serializing function

from xml.etree.ElementTree import parse
doc = parse(xml_str)

import binascii

s = b'hello'
h = binascii.b2a_hex(s) # bytes2hexbytes b'68656c6c6f'
b = binascii.a2b_hex(h) # hexbytes2bytes

import base64
s = b'hello'
h = base64.b16encode(s) # b'68656C6C6F' uppercase
b = base64.b16decode(h)

from struct import Struct

def write_records(records: tuple, format, f):
    record_struct = Struct(format)
    for r in records:
	f.write(record_struct.pack(*r))

def read_records(format, f) -> tuple:
    record_struct = Struct(format)
    chunks = iter(lambda: f.read(record_struct.size), b'') # star!
    return (record_struct.unpack(chunk) for chunk in chunks) # star!

def recv(maxsize, *, block):
    pass

recv(1024, True)# TypeError
recv(1024, block=True) # OK

x = 10
a = lambda y, x=x: x + y # use x=x to bind at definition time
x = 20
a(5) # => 15

_formats = {
    'ymd' : '{d.year}-{d.month}-{d.day}',
    'mdy' : '{d.month}/{d.day}/{d.year}',
    'dmy' : '{d.day}/{d.month}/{d.year}'
    }
from datetime import date
d = date.today()
format(d, 'mdy')
'The date is {:ymd}'.format(d)

class Connection:
    def __enter__(self):
	self.fs = open('somefile.txt', 'rt')
	return self.fs

    def __exit__(self, exc_ty, exc_val, tb):
	self.fs.close()
	self.fs = None

class Date:
    __slots__ = ['year', 'month', 'day']
    def __init__(self, year, month, day):
	self.year = year
	self.month = month
	self.day = day

class Object:
    @property
    def attr(self):
	return self.__attr

    @attr.setter
    def attr(self, value):
	self.__attr = value

    @attr.deleter
    def attr(self):
	self.__attr = None
    # del obj.attr

class String:
    def __init__(self, name):
	self.name = name

    def __get__(self, instance, cls):
	if instance is None:  # called as class variable
	    return self
	return instance.__dict__[self.name]

    def __set__(self, instance, value):
	if not isinstance(value, str):
	    raise TypeError('Expected a string')
	instance.__dict__[self.name] = value

    def __delete__(self, instance):
	del instance.__dict__[self.name]


class Person:
    name = String('name')

    def __init__(self, name):
	self.name = name

class Structure:
    # Class variable that specifies expected fields
    _fields = []

    def __init__(self, *args):
	if len(args) != len(self._fields):
	    raise TypeError('Expected {} arguments'.format(len(self._fields)))

	# Set the arguments
	for name, value in zip(self._fields, args):
	    setattr(self, name, value)


class Stock(Structure):
    _fields = ['name', 'shares', 'price']

def attach_wrapper(obj, func=None):
    if func is None:
	return partial(attach_wrapper, obj)
    setattr(obj, func.__name__, func)
    return func

import logging
from functools import partial, wraps


def logged(func=None, *, level=logging.DEBUG, name=None, message=None):
    if func is None:
	return partial(logged, level=level, name=name, message=message)
    logname = name if name else func.__module__
    log = logging.getLogger(logname)
    logmsg = message if message else func.__name__

    @wraps(func)
    def wrapper(*args, **kwargs):
	log.log(level, logmsg)
	return func(*args, **kwargs)

    return wrapper

from inspect import signature
from functools import wraps


def typeassert(*ty_args, **ty_kwargs):
    def decorate(func):
	# If in optimized mode, disable type checking
	if not __debug__:
	    return func
	# Map function argument names to supplied types
	sig = signature(func)
	bound_types = sig.bind_partial(*ty_args, **ty_kwargs).arguments

	@wraps(func)
	def wrapper(*args, **kwargs):
	    bound_values = sig.bind(*args, **kwargs)
	    # Enforce type assertions across supplied arguments
	    for name, value in bound_values.arguments.items():
		if name in bound_types:
		    if not isinstance(value, bound_types[name]):
			raise TypeError('Argument {} must be {}'.format(
			    name, bound_types[name]))
	    return func(*args, **kwargs)
	return wrapper
    return decorate


@typeassert(int, z=int)
def add(x, y, z=42):
    return x + y + z

import types
from functools import wraps


class Profiled:
    def __init__(self, func):
	wraps(func)(self)
	self.ncalls = 0

    def __call__(self, *args, **kwargs):
	self.ncalls += 1
	return self.__wrapped__(*args, **kwargs)

    def __get__(self, instance, cls):
	if instance is None:
	    return self
	else:
	    return types.MethodType(self, instance)


@Profiled
def add(x, y):
    return x + y


add(4, 5)
print(add.ncalls)

def patch(cls):
    orig_method = cls.method

    def new_method(self):
	return orig_method(self)

    cls.method = new_method

import operator

class StructTupleMeta(type):
    def __init__(cls, *args, **kwargs):
	super().__init__(*args, **kwargs)
	for n, name in enumerate(cls._fields):
	    setattr(cls, name, property(operator.itemgetter(n)))
	    # After f = itemgetter(2), the call f(r) returns r[2]

class StructTuple(tuple, metaclass=StructTupleMeta):
    _fields = []

    def __new__(cls, *args):
	if len(args) != len(cls._fields):
	    raise ValueError('{} arguments required'.format(len(cls._fields)))
	return super().__new__(cls, args)

class Stock(StructTuple):
    _fields = ['name', 'shares', 'price']

s = Stock('ACME', 50, 91.1)

def typed_property(name, expected_type):
    storage_name = '_' + name

    @property
    def prop(self):
	return getattr(self, storage_name)

    @prop.setter
    def prop(self, value):
	if not isinstance(value, expected_type):
	    raise TypeError('{} must be a {}'.format(name, expected_type))
	setattr(self, storage_name, value)
    return prop

from functools import partial
String = partial(typed_property, expected_type=str)
Integer = partial(typed_property, expected_type=int)

# Example use
class Person:
    name = String('name')
    age = Integer('age')

    def __init__(self, name, age):
	self.name = name
	self.age = age

import time
from contextlib import contextmanager

@contextmanager
def timethis(label):
    start = time.time()
    try:
	yield
    finally:
	end = time.time()
	print('{}: {}'.format(label, end - start))

# Example use
with timethis('counting'):
    n = 10000000
    while n > 0:
	n -= 1

class MyMeta(type):
    def __new__(self, clsname, bases, clsdict):
	# self is a class object
	# clsname is name of class being defined
	# bases is tuple of base classes
	# clsdict is class dictionary
	return super().__new__(self, clsname, bases, clsdict)

class MyMeta(type):
    def __init__(cls, clsname, bases, clsdict):
	super().__init__(clsname, bases, clsdict)

class NoInstances(type):
    def __call__(self, *args, **kwargs):
	raise TypeError("Can't instantiate directly")

# Example
class Spam(metaclass=NoInstances):
    @staticmethod
    def grok(x):
	print('Spam.grok')

class Singleton(type):
    def __init__(self, *args, **kwargs):
	self.__instance = None
	super().__init__(*args, **kwargs)

    def __call__(self, *args, **kwargs):
	if self.__instance is None:
	    self.__instance = super().__call__(*args, **kwargs)
	    return self.__instance
	else:
	    return self.__instance

import weakref

class Cached(type):
    def __init__(self, *args, **kwargs):
	super().__init__(*args, **kwargs)
	self.__cache = weakref.WeakValueDictionary()

    def __call__(self, *args):
	if args in self.__cache:
	    return self.__cache[args]
	else:
	    obj = super().__call__(*args)
	    self.__cache[args] = obj
	    return obj

# Metaclass that uses an OrderedDict for class body
class OrderedMeta(type):
    def __new__(cls, clsname, bases, clsdict):
	d = dict(clsdict)
	order = []
	for name, value in clsdict.items():
	    if isinstance(value, Typed):
		value._name = name
		order.append(name)
	d['_order'] = order
	return type.__new__(cls, clsname, bases, d)

    @classmethod
    def __prepare__(cls, clsname, bases):
	return OrderedDict()

class Spam(metaclass=MyMeta, debug=True, synchronize=True):
    ...

class MyMeta(type):
    # Optional
    @classmethod
    def __prepare__(cls, name, bases, *, debug=False, synchronize=False):
	# Custom processing
	...
	return super().__prepare__(name, bases)

    # Required
    def __new__(cls, name, bases, ns, *, debug=False, synchronize=False):
	# Custom processing
	...
	return super().__new__(cls, name, bases, ns)

    # Required
    def __init__(self, name, bases, ns, *, debug=False, synchronize=False):
	# Custom processing
	...
	super().__init__(name, bases, ns)

# __init__.py
def A():
    from .a import A
    return A()

# foo-package/
#   spam/
#     blah.py
# bar-package/
#   spam/
#     grok.py
import sys
sys.path.extend(['foo-package', 'bar-package'])
import spam.blah
import spam.grok

import ipaddress
net = ipaddress.ip_network('123.45.67.64/27')
list(net)
net.num_addresses

inet = ipaddress.ip_interface('123.45.67.73/27')
inet.network
inet.ip

def application(environ, start_response):
    start_response('200 OK', [('Content-Type', 'text/html')])
    return [b'<h1>Hello, web!</h1>']

from wsgiref.simple_server import make_server
httpd = make_server('', 8000, application)
print('Serving HTTP on port 8000...')
httpd.serve_forever()

from xmlrpc.server import SimpleXMLRPCServer

class KeyValueServer:
    _rpc_methods_ = ['get', 'set']
    def __init__(self, address):
	self._data = {}
	self._serv = SimpleXMLRPCServer(address, allow_none=True)
	for name in self._rpc_methods_:
	    self._serv.register_function(getattr(self, name))

    def get(self, name):
	return self._data[name]

    def set(self, name, value):
	self._data[name] = value

    def serve_forever(self):
	self._serv.serve_forever()

if __name__ == '__main__':
    kvserv = KeyValueServer(('', 15000))
    kvserv.serve_forever()

from xmlrpc.client import ServerProxy
s = ServerProxy('http://localhost:15000', allow_none=True)
s.set('foo', 'bar')
s.get('foo')

import hmac
import os

def client_authenticate(connection, secret_key):
    '''
    Authenticate client to a remote service.
    connection represents a network connection.
    secret_key is a key known only to both client/server.
    '''
    message = connection.recv(32)
    hash = hmac.new(secret_key, message)
    digest = hash.digest()
    connection.send(digest)


def server_authenticate(connection, secret_key):
    '''
    Request client authentication.
    '''
    message = os.urandom(32)
    connection.send(message)
    hash = hmac.new(secret_key, message)
    digest = hash.digest()
    response = connection.recv(len(digest))
    return hmac.compare_digest(digest, response)

import ssl
KEYFILE = 'server_key.pem'  # Private key of the server
CERTFILE = 'server_cert.pem'  # Server certificate (given to client)
s_ssl = ssl.wrap_socket(
    s, keyfile=KEYFILE, certfile=CERTFILE, server_side=True)

import select

def event_loop(handlers):
    while True:
	wants_recv = [h for h in handlers if h.wants_to_receive()]
	wants_send = [h for h in handlers if h.wants_to_send()]
	can_recv, can_send, _ = select.select(wants_recv, wants_send, [])
	for h in can_recv:
	    h.handle_receive()
	for h in can_send:
	    h.handle_send()

# zero copy with memoryview
def send_from(arr, dest):
    view = memoryview(arr).cast('B')
    while len(view):
	nsent = dest.send(view)
	view = view[nsent:]


def recv_into(arr, source):
    view = memoryview(arr).cast('B')
    while len(view):
	nrecv = source.recv_into(view)
	view = view[nrecv:]

import subprocess
out_bytes = subprocess.check_output(['netstat','-a'])
out_text = out_bytes.decode('utf-8')

import signal
import resource
import os


def time_exceeded(signo, frame):
    print("Time's up!")
    raise SystemExit(1)


def set_max_runtime(seconds):
    # Install the signal handler and set a resource limit
    soft, hard = resource.getrlimit(resource.RLIMIT_CPU)
    resource.setrlimit(resource.RLIMIT_CPU, (seconds, hard))
    signal.signal(signal.SIGXCPU, time_exceeded)

def limit_memory(maxsize):
    soft, hard = resource.getrlimit(resource.RLIMIT_AS)
    resource.setrlimit(resource.RLIMIT_AS, (maxsize, hard))

with self.assertRaisesRegex(ValueError, 'error*'):
    ...

import warnings
warnings.warn('logfile argument deprecated', DeprecationWarning)