Created
September 23, 2011 03:06
-
-
Save davidmarble/1236655 to your computer and use it in GitHub Desktop.
Some utility functions for django projects
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
""" | |
Some utilities for django projects | |
A few others: | |
https://github.com/andymccurdy/django-tips-and-tricks/blob/master/model_update.py | |
https://bitbucket.org/weholt/dse2 | |
""" | |
import sys, os, string, random | |
from django.contrib.contenttypes.models import ContentType | |
def constants(module, list_of_pairs): | |
current_module = sys.modules[module] | |
constants_inverted_dict = {} | |
for i, v in enumerate(list_of_pairs): | |
constants_inverted_dict.update({v[0]:v[1]}) | |
current_module.__dict__[v[1]] = v[0] | |
return (list_of_pairs, constants_inverted_dict) | |
def constants_and_strings(module, list_of_triples): | |
current_module = sys.modules[module] | |
list_of_pairs = [] | |
constants_inverted_dict = {} | |
constants_inverted_text_dict = {} | |
for i, v in enumerate(list_of_triples): | |
list_of_pairs.append((v[0], v[1])) | |
constants_inverted_dict.update({v[0]:v[1]}) | |
constants_inverted_text_dict.update({v[0]:v[2]}) | |
current_module.__dict__[v[1]] = v[0] | |
return (tuple(list_of_pairs), constants_inverted_dict, | |
constants_inverted_text_dict) | |
def makedirs(dirname): | |
if not os.path.isdir(dirname): | |
try: | |
os.makedirs(dirname) | |
except OSError: | |
# if os.path.exists(dirname): | |
if os.path.isdir(dirname): | |
# We are nearly safe | |
pass | |
else: | |
# There was an error on creation, so make sure we know about it | |
raise | |
def get_public_abspath_for_model(instance, filename): | |
ctype = ContentType.objects.get_for_model(instance) | |
model = ctype.model | |
app = ctype.app_label | |
filename = os.path.basename(filename) | |
if instance.pk: | |
return os.path.join("public", model, str(instance.pk), filename) | |
else: | |
return os.path.join("public", model, 'unfiled', filename) | |
def get_public_path_for_model(instance, filename): | |
ctype = ContentType.objects.get_for_model(instance) | |
model = ctype.model | |
app = ctype.app_label | |
filename = os.path.basename(filename) | |
if instance.pk: | |
return '/'.join(["public", model, str(instance.pk), filename]) | |
else: | |
return '/'.join(["public", model, 'unfiled', filename]) | |
def queryset_iterator(queryset, chunksize=1000): | |
''' | |
http://djangosnippets.org/snippets/1949/ | |
Iterate over a Django Queryset ordered by the primary key | |
This method loads a maximum of chunksize (default: 1000) rows in its | |
memory at the same time while django normally would load all rows in its | |
memory. Using the iterator() method only causes it to not preload all the | |
classes. | |
Note that the implementation of the iterator does not support ordered query sets. | |
''' | |
import gc | |
pk = 0 | |
last_pk = queryset.order_by('-pk')[0].pk | |
queryset = queryset.order_by('pk') | |
while pk < last_pk: | |
for row in queryset.filter(pk__gt=pk)[:chunksize]: | |
pk = row.pk | |
yield row | |
gc.collect() | |
# An alternative to the above | |
class MemorySavingQuerysetIterator(object): | |
""" | |
http://stackoverflow.com/questions/4856882/limiting-memory-use-in-a-large-django-queryset | |
""" | |
def __init__(self,queryset,max_obj_num=1000): | |
self._base_queryset = queryset | |
self._generator = self._setup() | |
self.max_obj_num = max_obj_num | |
def _setup(self): | |
for i in xrange(0,self._base_queryset.count(),self.max_obj_num): | |
# By making a copy of of the queryset and using that to actually access | |
# the objects we ensure that there are only `max_obj_num` objects in | |
# memory at any given time | |
smaller_queryset = copy.deepcopy(self._base_queryset)[i:i+self.max_obj_num] | |
logger.debug('Grabbing next %s objects from DB' % self.max_obj_num) | |
for obj in smaller_queryset.iterator(): | |
yield obj | |
def __iter__(self): | |
return self | |
def next(self): | |
return self._generator.next() | |
# This is from django-piston/piston/utils.py | |
def coerce_put_post(request): | |
""" | |
Django doesn't particularly understand REST. | |
In case we send data over PUT, Django won't | |
actually look at the data and load it. We need | |
to twist its arm here. | |
The try/except abominiation here is due to a bug | |
in mod_python. This should fix it. | |
""" | |
if request.method == "PUT": | |
# Bug fix: if _load_post_and_files has already been called, for | |
# example by middleware accessing request.POST, the below code to | |
# pretend the request is a POST instead of a PUT will be too late | |
# to make a difference. Also calling _load_post_and_files will result | |
# in the following exception: | |
# AttributeError: You cannot set the upload handlers after the upload has been processed. | |
# The fix is to check for the presence of the _post field which is set | |
# the first time _load_post_and_files is called (both by wsgi.py and | |
# modpython.py). If it's set, the request has to be 'reset' to redo | |
# the query value parsing in POST mode. | |
if hasattr(request, '_post'): | |
del request._post | |
del request._files | |
try: | |
request.method = "POST" | |
request._load_post_and_files() | |
request.method = "PUT" | |
except AttributeError: | |
request.META['REQUEST_METHOD'] = 'POST' | |
request._load_post_and_files() | |
request.META['REQUEST_METHOD'] = 'PUT' | |
request.PUT = request.POST | |
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment