Simply inherit from the list class, and override the __iter__ method. A great example, from SO:
import json
def gen():
yield 20
yield 30
yield 40
class StreamArray(list):
def __iter__(self):
return gen()
a = [1,2,3]
b = StreamArray()
print(json.dumps([1,a,b]))
I recently ran into an issue with the encoding of data coming back from MySQL through sqlalchemy. This is the first time that I’ve encountered such issues since this project first came online, months ago.
I am using utf8 encoding on my database, tables, and columns. I just added a new column, and suddenly my pages and/or AJAX calls started failing with one of the following two messages, respectively:
When I tell the stored procedure to return an empty string for the new column instead of its data, it works. The other text columns have an identical encoding.
It turns out that SQLAlchemy defaults to the latin1 encoding. If you need something different, than you’re in for a surprise. The official solution is to pass the “encoding” parameter to create_engine. This is the example from the documentation:
engine = create_engine("mysql://scott:tiger@hostname/dbname", encoding='latin1', echo=True)
In my case, I tried utf8. However, it still didn’t work. I don’t know if that ever works. It wasn’t until I uncovered a StackOverflow entry that I found the answer. I had to append “?charset=utf8” to the DSN string:
mysql+mysqldb://username:password@hostname:port/database_name?charset=utf8
The following are the potential explanations:
Whatever the case, it’s fixed and I’m a few hours older.
This is a sticky and exotic use-case of ctypes. In the example below, we make a call to some library function that treats ptr like a double-pointer, and sets ptr to point to a buffer and sets count with the number of bytes that are available there. The data at the pointer may have one or more NULL bytes that should not be interpreted as terminators.
from ctypes import *
ptr = ctypes.c_char_p()
count = ctypes.c_size_t()
r = library.some_call(
ctypes.cast(ctypes.byref(ptr),
ctypes.POINTER(ctypes.c_void_p)),
ctypes.byref(count))
if r != 0:
raise ValueError("Library call failed.")
data = ctypes.string_at(ptr, count.value)
Python 3.4 added a “singledispatch” decorator to functools, which provides method overloads. This enables you to perform different operations based on the type of the first argument.
By default, it prefers to work with static methods. This mostly comes from the link above:
import functools
class TestClass(object):
@functools.singledispatch
def test_method(arg):
print("Let me just say,", end=" ")
print(arg)
@test_method.register(int)
def _(arg):
print("Strength in numbers, eh?", end=" ")
print(arg)
@test_method.register(list)
def _(arg):
print("Enumerate this:")
for i, elem in enumerate(arg):
print(i, elem)
if __name__ == '__main__':
TestClass.test_method(55555)
TestClass.test_method([33, 22, 11])
However, there is a low-impact way to get overloading on instance-methods, too. We’ll just place our own wrapper around the standard singledispatch wrapper, and hijack the bulk of the functionality:
import functools
def instancemethod_dispatch(func):
dispatcher = functools.singledispatch(func)
def wrapper(*args, **kw):
return dispatcher.dispatch(args[1].__class__)(*args, **kw)
wrapper.register = dispatcher.register
functools.update_wrapper(wrapper, func)
return wrapper
class TestClass2(object):
@instancemethod_dispatch
def test_method(self, arg):
print("2: Let me just say,", end=" ")
print(arg)
@test_method.register(int)
def _(self, arg):
print("2: Strength in numbers, eh?", end=" ")
print(arg)
@test_method.register(list)
def _(self, arg):
print("2: Enumerate this:")
for i, elem in enumerate(arg):
print(i, elem)
if __name__ == '__main__':
t = TestClass2()
t.test_method(55555)
t.test_method([33, 22, 11])
Aside from superficial changes to the original example, we just added the instancemethod_dispatch function and updated the methods to take a “self” argument.
A special thanks to Zero Piraeus for penning the instancemethod_dispatch method (under the original name of “methdispatch”).
The use of print() is so commonplace and thoughtless that it’s easy to forget that it’s still a function. There are parameters often neglected. You may even find yourself using sys.stdout to avoid the automatic newline, which is folly.
This is the signature, as of 3.4:
print(*objects, sep=' ', end='\n', file=sys.stdout, flush=False)
None of the parameters need an explanation. The flush parameter was added in 3.3 .
Python 3.x introduced a relatively unknown feature called “function annotations”. This introduces a way to tag parameters and your return value with arbitrary information at the function-definition level.
You can annotate using strings or any other type that you’d like:
>>> def some_function(parm1: "Example parameter"):
... pass
...
>>> some_function.__annotations__
{'parm1': 'Example parameter'}
>>> x = 5
>>> def some_function_2(parm1: x * 20):
... pass
...
>>> some_function_2.__annotations__
{'parm1': 100}
You can also annotate the return:
>>> def some_function_3() -> 'return-value tag':
... pass
...
>>> some_function_3.__annotations__
{'return': 'return-value tag'}
It’s important to note that there are already strong conventions in how to document your parameters, thanks to Sphinx. Therefore, the utility of annotations will most likely be entirely in terms of functionality. For example, you can annotate closures on-the-fly:
import random
c_list = []
for i in range(10):
def closure_() -> random.random():
pass
c_list.append(closure_)
list(map(lambda x: print(x.__annotations__), c_list))
This is the output:
{'return': 0.9644971188983055}
{'return': 0.8639746158842893}
{'return': 0.18610468531065305}
{'return': 0.8528801446167985}
{'return': 0.3022338513329076}
{'return': 0.6455491244718428}
{'return': 0.09106740460937834}
{'return': 0.16987808849543917}
{'return': 0.9136478506241527}
{'return': 0.41691681086623544}
Absolutely nothing in the Python language or library is dependent on annotations, so they’re yours to play with or implement as you see fit.
S3 allows you to upload pieces of large files in parallel. Unfortunately, most/all of the examples that I’ve seen online are inefficient or inconvenient. For example:
None of these strategies hold a candle to Greenlets (running off different file-pointers to the same physical copy of the file).
This example is located at RandomUtility: s3_parallel.
This is the principal class. Go to the original source for the imports and the couple module-level constants.
class ParallelUpload(object):
def __init__(self, ak, sk, bucket_name, filepath,
chunk_size_b=_DEFAULT_CHUNK_SIZE_B,
monitor_interval_s=_DEFAULT_MONITOR_INTERVAL_S):
self.__ak = ak
self.__sk = sk
self.__bucket_name = bucket_name
self.__filepath = filepath
self.__s3_key_name = os.path.basename(filepath)
self.__chunk_size_b = chunk_size_b
self.__coverage = 0.0
self.__monitor_interval_s = _DEFAULT_MONITOR_INTERVAL_S
self.__filesize_b = os.path.getsize(self.__filepath)
self.__chunks = int(math.ceil(float(self.__filesize_b) /
float(self.__chunk_size_b)))
self.__progress = [0.0] * self.__chunks
def __get_bucket(self, bucket_name):
conn = boto.s3.connection.S3Connection(self.__ak, self.__sk)
return conn.lookup(bucket_name)
def __standard_upload(self):
bucket = self.__get_bucket(self.__bucket_name)
new_s3_item = bucket.new_key(self.__s3_key_name)
new_s3_item.set_contents_from_filename(
self.__filepath,
cb=self.__standard_cb,
num_cb=20)
def __standard_cb(self, current, total):
_logger.debug("Status: %.2f%%", float(current) / float(total) * 100.0)
def __multipart_cb(self, i, current, total):
self.__progress[i] = float(current) / float(total) * 100.0
def __transfer_part(self, (mp_info, i, offset)):
(mp_id, mp_key_name, mp_bucket_name) = mp_info
bucket = self.__get_bucket(mp_bucket_name)
mp = boto.s3.multipart.MultiPartUpload(bucket)
mp.key_name = mp_key_name
mp.id = mp_id
# At any given time, this will describe the farther percentage into the
# file that we're actively working on.
self.__coverage = max(
(float(offset) / float(self.__filesize_b) * 100.0),
self.__coverage)
# The last chunk might be shorter than the rest.
eff_chunk_size = min(offset + self.__chunk_size_b,
self.__filesize_b) - \
offset
with open(filepath, 'rb') as f:
f.seek(offset)
mp.upload_part_from_file(
f,
i + 1,
size=eff_chunk_size,
cb=functools.partial(self.__multipart_cb, i),
num_cb=100)
def __mp_show_progress(self):
while 1:
columns = [("%3d%% " % self.__progress[i])
for i
in range(self.__chunks)]
pline = ' '.join(columns)
_logger.debug(pline)
gevent.sleep(self.__monitor_interval_s)
def __multipart_upload(self):
bucket = self.__get_bucket(self.__bucket_name)
mp = bucket.initiate_multipart_upload(self.__s3_key_name)
mp_info = (mp.id, mp.key_name, mp.bucket_name)
chunk_list = range(0, self.__filesize_b, self.__chunk_size_b)
try:
gen = ((mp_info, i, offset)
for (i, offset)
in enumerate(chunk_list))
f = functools.partial(gevent.spawn, self.__transfer_part)
if self.__monitor_interval_s > 0:
p = gevent.spawn(self.__mp_show_progress)
g_list = map(f, gen)
gevent.joinall(g_list)
if self.__monitor_interval_s > 0:
p.kill()
p.join()
except:
mp.cancel_upload()
raise
else:
mp.complete_upload()
def start(self):
if self.__filesize_b < _MIN_MULTIPART_SIZE_B:
self.__standard_upload()
else:
self.__multipart_upload()
The output when called as a command will look like this:
$ python s3_parallel.py (access key) (secret key) (bucket name) (file-path) 2014-06-17 10:16:48,458 - __main__ - DEBUG - 0% 0% 0% 0% 0% 0% 0% 2014-06-17 10:16:58,459 - __main__ - DEBUG - 3% 3% 2% 2% 2% 1% 7% 2014-06-17 10:17:08,460 - __main__ - DEBUG - 6% 5% 5% 4% 5% 4% 14% 2014-06-17 10:17:18,461 - __main__ - DEBUG - 10% 7% 8% 8% 7% 6% 18% 2014-06-17 10:17:28,461 - __main__ - DEBUG - 16% 10% 13% 11% 10% 8% 26% 2014-06-17 10:17:38,462 - __main__ - DEBUG - 21% 14% 20% 15% 14% 12% 35% 2014-06-17 10:17:48,462 - __main__ - DEBUG - 26% 17% 27% 19% 19% 15% 48% 2014-06-17 10:17:58,463 - __main__ - DEBUG - 32% 20% 33% 24% 24% 18% 59% 2014-06-17 10:18:08,463 - __main__ - DEBUG - 37% 24% 39% 29% 28% 22% 70% 2014-06-17 10:18:18,464 - __main__ - DEBUG - 43% 28% 44% 34% 32% 26% 82% 2014-06-17 10:18:28,464 - __main__ - DEBUG - 48% 31% 50% 39% 36% 31% 91% 2014-06-17 10:18:38,465 - __main__ - DEBUG - 52% 35% 55% 44% 43% 36% 100% 2014-06-17 10:18:48,465 - __main__ - DEBUG - 60% 39% 63% 47% 47% 40% 100% 2014-06-17 10:18:58,466 - __main__ - DEBUG - 68% 44% 69% 53% 53% 45% 100% 2014-06-17 10:19:08,466 - __main__ - DEBUG - 77% 49% 75% 58% 57% 49% 100% 2014-06-17 10:19:18,467 - __main__ - DEBUG - 83% 54% 84% 65% 62% 52% 100% 2014-06-17 10:19:28,467 - __main__ - DEBUG - 88% 58% 90% 71% 69% 58% 100% 2014-06-17 10:19:38,468 - __main__ - DEBUG - 96% 61% 96% 77% 74% 63% 100% 2014-06-17 10:19:48,468 - __main__ - DEBUG - 100% 67% 100% 83% 83% 70% 100% 2014-06-17 10:19:58,469 - __main__ - DEBUG - 100% 73% 100% 93% 93% 76% 100% 2014-06-17 10:20:08,469 - __main__ - DEBUG - 100% 83% 100% 100% 100% 86% 100% 2014-06-17 10:20:18,470 - __main__ - DEBUG - 100% 95% 100% 100% 100% 100% 100%
Python 3.3 added a new file mode that allows you to create a new file and open it for write only if it does not already exist.
>>> with open('new_file', 'x') as f:
... pass
...
>>> with open('new_file', 'x') as f:
... pass
...
Traceback (most recent call last):
File "", line 1, in
FileExistsError: [Errno 17] File exists: 'new_file'
You might eventually have to unit-test a website that has a functional need to be run as SSL. For example, you might need to test a client that must connect using SSL authentication.
You can accomplish this by combining Python’s built-in webserver with ssl.SSLSocket.
This code is a distant relative of another example, but is lighter, simpler, and more Pythonic.
It runs out of the current directory (you’ll have to chdir() from the code if you want something different, since the webserver doesn’t take a path), and expects server.private_key.pem and server.crt.pem to exist.
import os.path
import socket
import SocketServer
import BaseHTTPServer
import SimpleHTTPServer
import ssl
class _SecureHTTPRequestHandler(SimpleHTTPServer.SimpleHTTPRequestHandler):
def setup(self):
self.connection = self.request
self.rfile = socket._fileobject(self.request, 'rb', self.rbufsize)
self.wfile = socket._fileobject(self.request, 'wb', self.wbufsize)
class _SecureHTTPServer(BaseHTTPServer.HTTPServer):
def __init__(self, private_key_pem_filepath, cert_pem_filepath,
binding=None, handler_cls=_SecureHTTPRequestHandler):
if binding is None:
# The default port is 1443 so that we don't have to be root.
binding = ('', 1443)
# We can't use super() because it's not a new-style class.
SocketServer.BaseServer.__init__(self, binding, handler_cls)
s = socket.socket(self.address_family, self.socket_type)
self.socket = ssl.SSLSocket(
s,
keyfile=private_key_pem_filepath,
certfile=cert_pem_filepath)
self.server_bind()
self.server_activate()
app_path = os.path.abspath(os.path.dirname(__file__))
private_key_pem_filepath = os.path.join(app_path, 'server.private_key.pem')
certificate_pem_filepath = os.path.join(app_path, 'server.crt.pem')
httpd = _SecureHTTPServer(
private_key_pem_filepath,
certificate_pem_filepath)
print("Running.")
httpd.serve_forever()
This code may also be found in the RandomUtility repository.
Use more than one for in the same list comprehension:
[(i, j, k) for i in (11, 22, 33) for j in (44, 55, 66) for k in (77, 88, 99)]
This results in all possible permutations (a cartesian product), where enumeration starts on the right:
[(11, 44, 77), (11, 44, 88), (11, 44, 99), (11, 55, 77), (11, 55, 88), (11, 55, 99), (11, 66, 77), (11, 66, 88), (11, 66, 99), (22, 44, 77), (22, 44, 88), (22, 44, 99), (22, 55, 77), (22, 55, 88), (22, 55, 99), (22, 66, 77), (22, 66, 88), (22, 66, 99), (33, 44, 77), (33, 44, 88), (33, 44, 99), (33, 55, 77), (33, 55, 88), (33, 55, 99), (33, 66, 77), (33, 66, 88), (33, 66, 99)]
You must be logged in to post a comment.