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
# Do the first 6 steps only once. | |
1. pip install --user alembic | |
2. bash: ``cd {{my_project}} && alembic init alembic`` | |
3. bash: ``text_editor {{my_project}}/alembic.ini`` | |
4. Change: "sqlalchemy.url = postgres://{{username}}:{{password}}@{{address}}/{{db_name}}" | |
5. bash: ``text_editor {{my_project}}/alembic/env.py`` | |
6. Now, import your metadata/db object from your app.: | |
# {{my_project}}/{{my_project_dir}}/app.py | |
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
# HANDLER | |
from django.utils.translation import ugettext_lazy as _ | |
from rest_framework.exceptions import ErrorDetail | |
from rest_framework.views import exception_handler | |
non_field_errors = 'non_field_errors' | |
def set_custome_errors(errors): |
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
# refactor of https://lukeplant.me.uk/blog/posts/double-checked-locking-with-django-orm/ | |
# untested | |
def double_checked_lock_iterator(queryset): | |
for item_pk in queryset.values_list("pk", flat=True): | |
with transaction.atomic(): | |
try: | |
yield queryset.select_for_update(skip_locked=True).get(id=item_pk) | |
except queryset.model.DoesNotExist: | |
pass |
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
class AfterFetchQuerySetMixin: | |
""" | |
QuerySet mixin to enable functions to run immediately | |
after records have been fetched from the DB. | |
""" | |
# This is most useful for registering 'prefetch_related' like operations | |
# or complex aggregations that need to be run after fetching, but while | |
# still allowing chaining of other QuerySet methods. | |
def __init__(self, *args, **kwargs): | |
super().__init__(*args, **kwargs) |
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
import logging | |
from concurrent.futures import ThreadPoolExecutor | |
from datetime import timedelta, datetime | |
from typing import Any, Dict | |
class SimpleStorageEntry: | |
_counter: int = 0 | |
id: int |
OlderNewer