Django

MPTT Tree ID Django + mysql： transaction.atomic() で select_for_update() を使ってレコードをロックしたときのメモ MySQL で採番機能（シーケンス）を実装する方法を整理する MySQL のロックについて公式ドキュメントを読みながら動作検証してみた〜テーブルレベルロック〜 分散ロック Distributed Locking in Django 1 2 3 4 from django.core.cache import cache with cache.lock("somekey"): do_some_thing() redis-py: https://github.com/redis/redis-py/blob/d3a3ada03e080f39144807c9fbe44876c40e0548/redis/client.py#L394 デッドロック Hook available for automatic retry after deadlock in django and mysql setup 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 import django.db.backends.utils from django.db import OperationalError import time original = django.db.backends.utils.CursorWrapper.execute def execute_wrapper(*args, **kwargs): attempts = 0 while attempts < 3: try: return original(*args, **kwargs) except OperationalError as e: code = e.args[0] if attempts == 2 or code != 1213: raise e attempts += 1 time.sleep(0.2) django.db.backends.utils.CursorWrapper.execute = execute_wrapper

MySQL: 𠮷（つちよし） データベース/テーブル utf8mb4にすること 1 ALTER TABLE customers_customer CONVERT TO CHARACTER SET utf8mb4 COLLATE utf8mb4_unicode_ci; 確認: $ echo "show variables like 'character%';" | python manage.py dbshell Variable_name Value character_set_client utf8mb4 character_set_connection utf8mb4 character_set_database utf8mb4 character_set_filesystem binary character_set_results utf8mb4 character_set_server utf8mb4 character_set_system utf8mb3 character_sets_dir /rdsdbbin/mysql-8.0.28.R4/share/charsets/ character_set_system utf8mb3 が問題 接続 を utf8mb4 にすること django: OPTIONS/charset=utf8mb4 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 In [1]: from django.conf import settings In [2]: settings.DATABASES Out[2]: {'default': {'ENGINE': 'django.db.backends.mysql', 'HOST': 'prod-db-instance.xxxxxxxx.ap-northeast-1.rds.amazonaws.com', 'NAME': 'coresys_masters', 'USER': 'coresys_masters', 'PASSWORD': 'va0Gaighoo3Paez8', 'OPTIONS': {'charset': 'utf8mb4', 'init_command': "SET sql_mode='STRICT_TRANS_TABLES'"}, 'ATOMIC_REQUESTS': False, 'AUTOCOMMIT': True, 'CONN_MAX_AGE': 0, 'TIME_ZONE': None, 'PORT': '', 'TEST': {'CHARSET': None, 'COLLATION': None, 'MIGRATE': True, 'MIRROR': None, 'NAME': None}}}

Django : 動的モデル生成 定義されたメタ情報で同じテーブルを複数作る 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 from pathlib import Path import pandas as pd from django.conf import settings from django.db import connection, models def get_meta(name): """ 定義情報(CSV)ファイル """ path = META_DIR / f"{name}.map.csv" return pd.read_csv(path, dtype="str").fillna("").to_dict(orient="records") def named_model(name, code): meta = get_meta(name) class_name = f"{name}{code}" table_name = f"{name}_{code}" unique_fields = list(map(lambda i: i["column"], filter(lambda i: i["unique"] == "1", meta))) class Meta: db_table = f"oracle_{table_name}" constraints = [ models.UniqueConstraint( fields=unique_fields, name=f"uniq_{table_name}", ), ] def _field(item): params = {"db_index": True} if item["db_index"] else {} params["blank"] = True return ( item["column"], models.CharField(verbose_name=item["label"], max_length=100, null=True, default=None, **params), ) fields = dict(map(_field, meta)) attrs = {"__module__": "oracle.models", "Meta": Meta, **fields} models_class = type(class_name, (models.Model,), attrs) return models_class def get_named_model(name, code): # モデルクラス取得 model_class = named_model(name, code) if model_class._meta.db_table in connection.introspection.table_names(): # テーブルが存在していたら終了 return model_class # テーブルの作成 with connection.schema_editor() as schema_editor: schema_editor.create_model(model_class) return model_class

Django : マイグレーション: Unique フィールドの追加 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 # Generated by Django 3.2.20 on 2023-07-24 06:36 from django.db import migrations, models def advise_fill_advise_code(apps, schema_editor): """ instance.id => instance.advise_code と初期値にする""" Model = apps.get_model("studies", "Advise") for instance in Model.objects.all(): instance.advise_code = str(instance.id) instance.save() class Migration(migrations.Migration): dependencies = [ ("studies", "0015_auto_20211108_1814"), ] operations = [ migrations.AddField( model_name="advise", name="advise_code", field=models.CharField( # オリジナル: # help_text="(id)", max_length=6, unique=True, verbose_name="Advise Code" # 修正: null=True default=None とする help_text="(id)", max_length=6, null=True, default=None, unique=True, verbose_name="Advise Code" ), preserve_default=False, ), # 追加: advise_code (ユニーク) を設定する migrations.RunPython(inspadvise_fill_advise_code, reverse_code=migrations.RunPython.noop), # 追加: null=True default=None を抜く migrations.AlterField( model_name="advise", name="advise_code", field=models.CharField(help_text="(id)", max_length=6, unique=True, verbose_name="Advise Code"), preserve_default=False, ), ]

Elasic Cache for Redis バージョン 1 2 3 4 5 $ redis-cli --version redis-cli 5.0.7 $ redis-server -v Redis server v=5.0.7 sha=00000000:0 malloc=jemalloc-5.2.1 bits=64 build=66bd629f924ac924 接続: 1 REDIS_URL=redis://mycloud-stage-redis-nc.sigrvp.ng.0001.apne1.cache.amazonaws.com:6379/0 1 2 3 $ export $(cat .env|xargs) $ redis-cli -u $REDIS_URL PING PONG Python redis-py: https://github.com/redis/redis-py django-redis: https://github.com/jazzband/django-redis channels_redis: https://github.com/django/channels_redis クラスターモードへの接続 Cluster Client 1 2 3 4 5 6 7 8 9 import logging from redis.cluster import RedisCluster ENDPOINT = "mycloud-stage-redis-test.sigrvp.clustercfg.apne1.cache.amazonaws.com" logging.basicConfig(level=logging.INFO) redis = RedisCluster(host=ENDPOINT, port="6379") if redis.ping(): logging.info("Connected to Redis") django-redis: ...

Celery on ECS 【Python Celery】本番運用時に知っておくべき 10 のこと https://github.com/Alexmhack/Django-Celery-Redis-AWSEB Redis Celery + AWS Redis で使ってみる クラスターモード Redis >= 3.2 ElastiCache for Redis のクラスターモードについて調べてみる SQS broker としてつかえるが、 backend としては使えない(SQS + RabbitMQ/Redis/SQLAlchemy を考える) Celery + SQS on Django on Elastic Beanstalk Deploying Django on AWS: Setting up Celery and SQS Deploying Django Application on AWS with Terraform. Setting up Celery and SQS ECS running celery tasks and celery beat in ECS with Django docker compose ecs deploy aws tutorial: celery incompatible attribute #8040 AWS Batch を検討したけど AWS Fargate を採用した話 Celery Logging celery.app.log Three Ideas to Customise Celery logging handlers 標準出力: ...

django: cache value decorator 1 2 from functools import wraps from django.core.cache import c 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 def cached_value(key=None, **kwargs): """ key で指定された値をキャッシュから取得して current に渡す reset=True が指定されるとキャッシュデータを使わない """ def _cache_controller(original_func): @wraps(original_func) def _cache_controlled(*args, **kw): current = None if kw.get("reset", False) else cache.get(key) lastest = original_func(*args, current=current, **kw) cache.set(key, lastest) return lastest return _cache_controlled return _cache_controller 1 2 3 4 5 class PurchaseOrder: @cached_value(key="webservice_accesstoken") def get_accesstoken(self, current=None, reset=False): return Token(self.company.credentials).update_token(current)

django-import-export M2M, FK django-import-export, importing ManyToMany from XLSX Widgets GroupWidget どちらでも使える: Users.groups Permission.group_set 1 2 3 4 5 6 7 8 9 10 11 12 from import_export import widgets, fields class GroupWidget(widgets.ManyToManyWidget): def clean(self, value, row=None, *args, **kwargs): if not value: return self.model.objects.none() if self.field == "name": # 存在しない場合作成する ids = value.split(self.separator) ids = filter(None, [i.strip() for i in ids]) return map(lambda i: self.model.objects.get_or_create(name=i)[0], ids) return super().clean(self, value, row=row, *args, **kwargs) ContentTypeWidget Permission.content_type など ForeingKey 定義されていると使える データは {app_label}.{model} でレンダリングされるので、取り込む時に split('.') する 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 from import_export import fields, widgets from django.contrib.contenttypes.models import ContentType class ContentTypeWidget(widgets.ForeignKeyWidget): def clean(self, value, row=None, *args, **kwargs): if value is None: return None app_label, model = value.split(".") content_type = ContentType.objects.get_by_natural_key(app_label, model) return content_type def render(self, value, obj=None): return f"{value.app_label}.{value.model}" User.groups User の Resource を定義するにあたって、groups を name で import/export するには、以下のように記述することができます。 ...

Django: 指定したグループに権限を与える 1 2 3 4 from itertools import product from django.contrib.auth.models import Permission, Group from django.contrib.contenttypes.models import ContentType from django.apps import apps 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 @main.command() @click.argument("group_name") @click.argument("model_path") @click.argument("action_suffix", nargs=-1) # view とか view_other とか @click.pass_context def group_add_perms(ctx, group_name, model_path, action_suffix): """指定したグループに権限を与える""" group = Group.objects.get(name=group_name) app_label, model_name = model_path.split(".") if app_label == "*": model_list = apps.get_models() else: model_list = list( filter( lambda i: model_name == "*" or i._meta.model_mame == model_name, apps.get_app_config(app_label).get_models(), ) ) actions = map(lambda i: i.split("_"), action_suffix) def _perm(item): model_class, actions = item content_type = ContentType.objects.get_for_model(model_class) codename = ( f"{actions[0]}_{content_type.model}_{actions[1]}" if len(actions) > 1 else f"{actions[0]}_{content_type.model}" ) return Permission.objects.filter(content_type=content_type, codename=codename).first() perms = filter(lambda i: i is not None, map(_perm, product(model_list, actions))) group.permissions.add(*perms)

django-import-export で 銀行マスターを取り込む 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 from import_export import resources from .. import models class BankCodeResource(resources.ModelResource): class Meta: model = models.BankCode import_id_fields = ["code", "branch_code"] fields = ["code", "branch_code", "kana", "name", "branch_kana", "branch_name"] def before_import_row(self, row, row_number=None, **kwargs): for field in self._meta.fields: # ヘッダーは verbose_name に合わせている value = row.pop(self._meta.model._meta.get_field(field).verbose_name, None) if value is None: continue # コードは0埋めする if field == "code": value = str(value).zfill(4) elif field == "branch_code": value = str(value).zfill(3) row[field] = value