Here are set of internal tools that are shared between different projects internally. Originally most tools related to testing, so they provide some base classes for various cases in testing
NOTE: all our tools are intentially support only 3.8+ python.
Some might work with other versions, but we're going to be free from all these crutches to backport things like async/await to lower versions, so if it works - fine, if not - feel free to send PR, but it isn't going to be merged all times.
# cache async function that returns pydantic.BaseModel
from fan_tools.python import cache_async
@cache_async[type(dict)](fname, model, {})
async def func():
return model
# cache sync function that returns json serializable response
from fan_tools.python import memoize
@memoize
def func(*args, **kwargs):
return json.dumps(
{
'args': args,
'kwargs': kwargs,
}
)
Defined in fan_tools/testing/__init__.py. Required for defining nested urls with formatting.
You can use it in fixtures, like:
@pytest.fixture(scope='session')
def api(api_v_base):
yield ApiUrls('{}/'.format(api_v_base), {
'password_reset_request': 'password/request/code/',
'password_reset': 'password/reset/',
'user_review_list': 'user/{user_id}/review/',
'user_review': 'user/{user_id}/review/{review_id}/',
'wine_review': 'wine/{wine_id}/review/',
'drink_review': 'drink/{drink_id}/review/',
})
def test_review_list(user, api):
resp = user.get_json(api.user_review_list(user_id=user1.id), {'page_size': 2})
You can find source in fan_tools/testing/meta.py.
For now it convert methods that are started with prop__ into descriptors with cache.
class A(metaclass=PropsMeta):
def prop__conn(self):
conn = SomeConnection()
return conn
Became:
class A:
@property
def conn(self):
if not hasattr(self, '__conn'):
setattr(self, '__conn', SomeConnection())
return self.__conn
Thus it allows quite nice style of testing with lazy initialization. Like:
class MyTest(TestCase, metaclass=PropsMeta):
def prop__conn(self):
return psycopg2.connect('')
def prop__cursor(self):
return self.conn.cursor()
def test_simple_query(self):
self.cursor.execute('select 1;')
row = self.cursor.fetchone()
assert row[0] == 1, 'Row: {}'.format(row)
Here you just get and use self.cursor, but automatically you get connection and cursor and cache they.
This is just simple example, complex tests can use more deep relations in tests. And this approach is way more easier and faster than complex setUp methods.
Basic unix helpers
run with return code and stderr checkwait_socket('localhost', 5432)succ for use with await. supports realtime loggingFormat string with system variables + defaults.
PG_DEFAULTS = {
'PGDATABASE': 'postgres',
'PGPORT': 5432,
'PGHOST': 'localhost',
'PGUSER': 'postgres',
'PGPASSWORD': '',
}
DSN = interpolate_sysenv('postgresql://{PGUSER}:{PGPASSWORD}@{PGHOST}:{PGPORT}/{PGDATABASE}', PG_DEFAULTS)
Enable json output with additional fields, suitable for structured logging into ELK or similar solutions.
Accepts env_vars key with environmental keys that should be included into log.
# this example uses safe_logger as handler (pip install safe_logger)
import logging
import logging.config
LOGGING = {
'version': 1,
'disable_existing_loggers': True,
'formatters': {
'json': {
'()': 'fan_tools.fan_logging.JSFormatter',
'env_vars': ['HOME'],
},
'standard': {
'format': '%(asctime)s [%(levelname)s] %(name)s: %(message)s'
},
},
'handlers': {
'default': {
'level': 'DEBUG',
'class': 'safe_logger.TimedRotatingFileHandlerSafe',
'filename': 'test_json.log',
'when': 'midnight',
'interval': 1,
'backupCount': 30,
'formatter': 'json',
},
},
'loggers': {
'': {
'handlers': ['default'],
'level': 'DEBUG',
},
},
}
logging.config.dictConfig(LOGGING)
log = logging.getLogger('TestLogger')
log.debug('test debug')
log.info('test info')
log.warn('test warn')
log.error('test error')
FastAPI based server that servers metrics in prometheus format.
import uvicorn
from fan_tools.mon_server.certs import update_certs_loop
app = FastAPI()
mserver = MetricsServer(app)
mserver.add_task(update_certs_loop, hosts=['perfectlabel.io', 'robopickles.com'])
uvicorn.run(app, host='0.0.0.0', port=os.environ.get('MONITORING_PORT', 8000))
There are two backup helpers: fan_tools.backup.s3.S3backup and fan_tools.backup.gcloud.GCloud
We're assuming that backup script has access to backup execution and dump directory.
Default setup includes support for docker container that access DB.
By default script provides interface for monitoring (last backup date).
fan_tools.backup.s3.S3backup provides external script called fan_s3_backup that has accepts some configuration via environmental variables.
false value, default: false/create_backup.py-b/--bucket - to define bucket. default for s3: environmental variable AWS_BACKUP_BUCKET-p/--prefix - directory backup prefix, usually it is subfolder for dumps, default: backups/-d/--daemonize - should we run in daemonized modeS3 specific:
Allow you to deserealize incoming strings into Enum values.
You should add EnumSerializer into your serializers by hand.
from enum import IntEnum
from django.db import models
from rest_framework import serializers
from fan_tools.drf.serializers import EnumSerializer
class MyEnum(IntEnum):
one = 1
two = 2
class ExampleModel(models.Model):
value = models.IntegerField(choices=[(x.name, x.value) for x in MyEnum])
class ExampleSerializer(serializers.ModelSerializer):
value = EnumSerializer(MyEnum)
# this allows you to post value as: {'value': 'one'}
Due to Enum and IntegerField realizations you may use Enum.value in querysets
ExampleModel.objects.filter(value=MyEnum.two)
LoggerMiddleware will log request meta + raw post data into log.
For django<1.10 please use fan_tools.django.log_requests.DeprecatedLoggerMiddleware
Decorator adds a unique for each uwsgi request dict as first function
argument.
For tests mock _get_request_unique_cache
Make function called only once on transaction commit. Here is examples
where function do_some_useful will be called only once after
transaction has been committed.
class SomeModel(models.Model):
name = IntegerField()
@call_once_on_commit
def do_some_useful():
pass
def hook(sender, instance, **kwargs):
do_some_useful()
models.signals.post_save.connect(hook, sender=SomeModel)
with transaction.atomic():
some_model = SomeModel()
some_model.name = 'One'
some_model.save()
some_model.name = 'Two'
some_model.save()
For tests with nested transactions (commit actually most times is not
called) it is useful to override behaviour call_once_on_commit
when decorated function executed right in place where it is called.
To do so mock on_commit function. Example pytest fixture:
@pytest.fixture(scope='session', autouse=True)
def immediate_on_commit():
def side_effect():
return lambda f: f()
with mock.patch('fan_tools.django.on_commit', side_effect=side_effect) as m:
yield m
Used for choices attribute for in model field
class FooBarEnum(ChoicesEnum):
foo = 1
bar = 2
class ExampleModel(models.Model):
type = models.IntegerField(choices=FooBarEnum.get_choices())
Allow to set postgres trigram word similarity threshold for default django database connection
set_word_similarity_threshold(0.4)
Django Model containing postgres ltree
class LTreeExampleModel(LTreeModel):
Lookup for postgres ltree descendants
LTreeExampleModel.objects.filter(path__descendants='root.level1')
Lookup for postgres ltree by level depth
LTreeExampleModel.objects.filter(path__nlevel=2)
Postgres text %> text operator
# Add this import to models.py (file should be imported before lookup usage)
import fan_tools.django.db.pgfields # noqa
Books.objects.filter(title__similar='Animal Farm')
Postgres text1 <<-> text2 operator. It returns 1 - word_similarity(text1, text2)
from django.db.models import Value, F
similarity = WordSimilarity(Value('Animal Farm'), F('title'))
Books.objects.annotate(similarity=similarity)
Django filter that match if integer is in the integers list separated by comma
class ExampleFilterSet(FilterSet):
example_values = NumberInFilter(field_name='example_value', lookup_expr='in')
Send text and html emails using django templates.
Mail(
recipient_list=[user.email],
template_name='user/emails/reset_password',
context={
'frontend_url': settings.FRONTEND_URL,
},
).send()
Get domain section of absolute url of current page using django request object.
build_absolute_uri(request)
Helps to use power of serializers for simple APIs checks.
from rest_framework import serializers
from rest_framework.decorators import api_view
from fan_tools.drf import use_form
class SimpleForm(serializers.Serializer):
test_int = serializers.IntegerField()
test_str = serializers.CharField()
@api_view(['GET'])
@use_form(SimpleForm)
def my_api(data):
print(f'Data: {data["test_int"]} and {data["test_str"]}')
Allow turn off pagination by specifying zero page_zize.
REST_FRAMEWORK = {
'DEFAULT_PAGINATION_CLASS': 'fan_tools.drf.pagination.ApiPageNumberPagination',
...
}
Pretty Django Rest Framework API renderer with error codes.
REST_FRAMEWORK = {
'DEFAULT_RENDERER_CLASSES': (
'fan_tools.drf.renderers.ApiRenderer',
},
...
}
Pretty Django Rest Framework API exception handler with error codes.
REST_FRAMEWORK = {
'EXCEPTION_HANDLER': 'fan_tools.drf.handlers.api_exception_handler',
...
}
Helper assert function to be used in tests to match the validation error codes.
assert_validation_error(response, 'email', 'unique')
Asyncio worker which wait for new records in postgres db table and process them.
aiopg shortcuts
Backport of python's 2 execfile function.
Usage: execfile('path/to/file.py', globals(), locals())
Returns: True if file exists and executed, False if file doesn't exist
Sphinx extensions to generate documentation for django restframework serializers and examples for http requests.
In order to use them specify dependency for package installation:
pip install fan_tools[doc_utils]
Usage:
# Add to Sphinx conf.py
extensions = [
# ...
'fan_tools.doc_utils.fan_sphinx.dyn_serializer',
'fan_tools.doc_utils.fan_sphinx.http_log'
]
Convert template yaml with substituion of %{ENV_NAME} strings to appropriate environment variables.
Usage: fan_env_yaml src_file dst_file
Helper to run default CI pipeline. Defaults are set up for giltab defaults. Includes stages:
It's optimized for parallel launches, so you need to use unique temporary name (--temp-name). We want keep our system clean if possible, so we'll delete this tag in the end. But we don't want to repeat basic steps over and over, so we will cache image with common cache name (--cache-name), it will remove previous cached image.
Wait for socket awailable/not-awailable with timeout.
# Wait until database port up for 180 seconds
fan_wait -t 180 postgres 5432
# Wait until nginx port down for 30 seconds
fan_wait -t 30 nginx 80
-e KEY=VALUE -e KEY2=VALUE2fan_env_yaml {TEMPLATE} /tmp/filebeat.yml/usr/bin/filebeat /tmp/filebeat.ymlrun_filebeat -e CHECKME=VALUE path_to_template
usage: doc_serializer [-h] [--rst] [--artifacts]
Parse serializers sources
optional arguments:
-h, --help show this help message and exit
--rst Output rst with serializers
--artifacts Write serializers artifacts
Save rotated by exif tag images. Some browsers/applications don't respect this tag, so it is easier to do that explicitly.
class Image(models.Model):
uploaded_by = models.ForeignKey(User, blank=True, null=True, on_delete=models.SET_NULL)
image = models.ImageField(blank=True, upload_to=image_upload_to)
thumb_image = models.ImageField(blank=True, upload_to=thumb_upload_to)
full_url = models.CharField(blank=True, max_length=255)
thumb_url = models.CharField(blank=True, max_length=255)
created = models.DateTimeField(auto_now_add=True)
updated = models.DateTimeField(auto_now=True)
class ImageSerializer(ModelSerializer):
class Meta:
model = Image
fields = ['id', 'created', 'updated', 'full_url', 'thumb_url']
class UploadImageView(views.GenericAPIView):
permission_classes = [IsAuthenticated]
def post(self, request, *args, **kwargs):
image_data = request.data['image']
# Fix an image orientation based on exif and remove exif from the resulted image.
transformed_image = Transpose().process(image_data)
obj = Image.objects.create(uploaded_by=request.user, image=transformed_image)
obj.full_url = obj.image.url
obj.save()
s = ImageSerializer(instance=obj)
return Response(s.data)
Helper to send metrics. Example for datadog
Usually you want to setup some kind of notification for metric with name error_metric. It is sent by send_error_metric.
For DataDog your metric query will look like:
sum:error_metric{service:prod*} by {error_type,service}.as_count()
# keep docker container
tox -e py311-django40 -- --keep-db django_tests
tox -e py311-django40 -- --keep-db --docker-skip django_tests
FAQs
Various python stuff: testing, aio helpers, etc
We found that fan-tools demonstrated a healthy version release cadence and project activity because the last version was released less than a year ago. It has 1 open source maintainer collaborating on the project.
Did you know?
Socket for GitHub automatically highlights issues in each pull request and monitors the health of all your open source dependencies. Discover the contents of your packages and block harmful activity before you install or update your dependencies.
Security News
PyPI confirms no security flaws were exploited in the Ultralytics supply chain attack and highlights improvements for safer package publishing.
Security News
Research
The Socket Research Team breaks down a malicious wrapper package that uses obfuscation to harvest credentials and exfiltrate sensitive data.
Research
Security News
Attackers used a malicious npm package typosquatting a popular ESLint plugin to steal sensitive data, execute commands, and exploit developer systems.