Commit cf826d8d authored by Nicolas Noirbent's avatar Nicolas Noirbent

Initial code commit

parent 2bbab965
__import__('pkg_resources').declare_namespace(__name__)
# -*- coding: utf-8 -*-
import os
import sys
import atexit
from .logging import get_logger
from django.core.management.base import BaseCommand
from django.conf import settings
class HexACKBaseCommand(BaseCommand):
"""
Extend Django BaseCommand with the following features:
- pid management
- logging
"""
def __init__(self, pid_n=None):
super().__init__()
self.command_name = self.get_command_name()
if pid_n is not None:
pid_name = pid_n
else:
pid_name = self.command_name
self.logger = get_logger(self.command_name)
self.manage_pid(pid_name)
def manage_pid(self, pid_name):
self.pidfile = os.path.join(settings.PID_DIR, pid_name+'.pid')
# no concurrent batch!
if os.path.exists(self.pidfile):
pf = open(self.pidfile, 'r')
pid = pf.read().strip()
pf.close()
self.logger.critical('Another batch is running: ' + pid)
sys.exit(1)
# create fullzap pid directory
if not os.path.exists(settings.PID_DIR):
os.makedirs(settings.PID_DIR)
pid = os.getpid()
atexit.register(self.delpid)
open(self.pidfile, 'w+').write("%s\n" % pid)
def delpid(self):
os.remove(self.pidfile)
def get_command_name(self):
return os.path.splitext(sys.argv[1])[0]
'''
Custom DRF / django-filter fields and filters.
'''
import datetime
import math
from django.core.exceptions import ValidationError
from django.forms.fields import FloatField
from django.utils import timezone
import django_filters
class TimestampField(FloatField):
def to_python(self, value):
value = super().to_python(value)
if value is None:
return value
if not math.isfinite(value):
raise ValidationError(self.error_messages['invalid'], code='invalid')
if not value >= 0:
raise ValidationError('Enter a positive number.', code='positive')
value = datetime.datetime.fromtimestamp(value, tz=timezone.utc)
return value
def validate(self, value):
if value in self.empty_values and self.required:
raise ValidationError(self.error_messages['required'], code='required')
class TimestampFilter(django_filters.NumberFilter):
field_class = TimestampField
'''
Custom DRF serializer fields
'''
import datetime
from rest_framework import serializers
class TimestampField(serializers.DateTimeField):
def to_representation(self, value):
return value.timestamp()
def to_internal_value(self, value):
# FIXME: float validation etc.
value = float(value)
result = datetime.datetime.fromtimestamp(value, tz=datetime.timezone.utc)
# The field supports passing datetime objects as input, so we let DRF do any additional validation it may need.
return super().to_internal_value(result)
# -*- coding: utf-8 -*-
import os
from PIL import Image
def resize_image(path_src, resolution, suffix, dir_dst=None):
image = Image.open(path_src)
if image.mode not in ('L', 'RGB'):
image = image.convert('RGB')
min_width = float(resolution[0])
min_height = float(resolution[1])
# Keep image aspect ratio
width_ratio = image.size[0]/min_width
height_ratio = image.size[1]/min_height
if width_ratio > height_ratio:
resolution = (int(min_width), int(image.size[1]/width_ratio))
else:
resolution = (int(image.size[0]/height_ratio), int(min_height))
image_resized = image.resize(resolution, Image.ANTIALIAS)
filename_src = os.path.basename(path_src)
filename_dst = (suffix).join(os.path.splitext(filename_src))
path_dst = os.path.join(os.path.dirname(path_src), filename_dst)
if dir_dst and os.path.exists(dir_dst) and os.path.isdir(dir_dst):
path_dst = os.path.join(dir_dst, filename_dst)
image_resized.save(path_dst, image.format, quality=85)
return path_dst
# -*- coding: utf-8 -*-
import logging
import os
from django.conf import settings
def get_logger(log_name):
os.makedirs(settings.LOG_DIR, exist_ok=True)
logfile_handler = logging.handlers.TimedRotatingFileHandler(settings.LOG_DIR + "/" + log_name,
when="midnight",
interval=1,
backupCount=30)
formatter = logging.Formatter('%(asctime)s [%(levelname)s] %(message)s', '%Y-%m-%d %H:%M:%S')
logfile_handler.setFormatter(formatter)
logfile_handler.suffix = "%Y-%m-%d"
# Critical errors go to stderr, useful for cron commands
console_handler = logging.StreamHandler()
console_handler.setLevel(logging.CRITICAL)
logger = logging.getLogger(log_name)
if not logger.handlers:
logger.addHandler(logfile_handler)
logger.addHandler(console_handler)
if settings.DEBUG:
logger.setLevel(logging.DEBUG)
else:
logger.setLevel(logging.INFO)
return logger
#! /usr/bin/env python
'''
This file is needed so Django can properly find and import the tests.settings module.
'''
import os
import sys
from django.core.management import execute_from_command_line
def runtests():
os.environ.setdefault('DJANGO_SETTINGS_MODULE', 'tests.settings')
return execute_from_command_line(sys.argv)
if __name__ == '__main__':
sys.exit(runtests())
#! /usr/bin/env python3
# -*- coding: utf-8 -*-
from setuptools import setup, find_packages
setup(
name='hexack-utils',
version="0.1.0dev",
packages=find_packages(exclude=['tests']),
namespace_packages=['hexack'],
install_requires=[
'Pillow',
'django>=1.1',
'django-filter>=2',
'djangorestframework>=3'
],
python_requires='>=3.4',
author='HexACK',
author_email='dev@hexack.fr',
description='HexACK common utils',
url='https://github.com/hexack/hexack-utils',
license='MIT',
)
SECRET_KEY = 'test key'
INSTALLED_APPS = (
'django.contrib.auth',
'django.contrib.contenttypes',
'tests',
'rest_framework',
)
USE_TZ = True
MIDDLEWARE = []
DATABASES = {
'default': {
'ENGINE': 'django.db.backends.sqlite3',
'NAME': ':memory:',
}
}
import datetime
import unittest
from hexack.utils.drf.serializers import TimestampField as SerializerTimestampField
class SerializerTimestampFieldTestCase(unittest.TestCase):
def test_representation(self):
field = SerializerTimestampField()
test_date = datetime.datetime(2019, 1, 18, 12, 34, 56, tzinfo=datetime.timezone.utc)
self.assertEqual(field.to_representation(test_date), 1547814896.0)
def test_internal_value(self):
field = SerializerTimestampField()
test_timestamp = 1547814896.0
self.assertEqual(field.to_internal_value(test_timestamp),
datetime.datetime(2019, 1, 18, 12, 34, 56, tzinfo=datetime.timezone.utc))
[tox]
envlist =
py34
py35
py36
py37
flake8
[testenv]
deps = coverage
commands =
coverage run --source=hexack ./runtests.py test {posargs}
coverage report -m
coverage html
setenv =
DJANGO_SETTINGS_MODULE=tests.settings
[testenv:flake8]
deps = flake8
commands = flake8 --count
[flake8]
max-line-length = 120
exclude =
.tox
.eggs
Markdown is supported
0% or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment