Web開発

With 【Python】with 文の構造を理解する

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

UTF8 -> CP932 【pandas】csv 出力時に cp932 を指定すると UnicodeError が起こる場合の対処法 [Python3] Shift_JIS と UTF-8 と ASCII を行き来する コーデック: エラーハンドラ UTF-8 → cp932(Shift_JIS)変換表

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, ), ]

patch.object(): メソッドを置き換える メソッドが外部のサーバーに対して API リクエストを行う テストではサーバー間 API が成功したものとして、状態だけ進める 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 from unittest.mock import patch from applies.utils import approve_to from base.tests.utils import RestTestCase from sourcing.models import PurchaseOrder def fake_cloudsign_post_document(self, *args, **kwargs): # サーバー間のAPIが成功したものとして承認状態を進める approve_to(self.purchase, "purchase_print") class PurchaseOrderTest(RestTestCase): @patch.object(PurchaseOrder, "cloudsign_post_document", new=fake_cloudsign_post_document) def test_create(self): # .... # response = client.post("/api/rest/sourcing/purchaseorder/", params) ... instance = PurchaseOrder.objects.get(id=response.json()["id]) self.assertEqual(instanc.purchase.purchase_status, "print")

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)

at コマンド MacOS で at コマンドを有効化して使ってみる 手順: /System/Library/LaunchDaemons/com.apple.atrun.plist で、Disabled を false に変更 /System/Library/LaunchDaemons/com.apple.atrun.plist を再ロード 1 2 3 4 % sudo launchctl load -w /System/Library/LaunchDaemons/com.apple.atrun.plist /System/Library/LaunchDaemons/com.apple.atrun.plist: service already loaded Load failed: 37: Operation already in progress

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)