接続先のインスタンスを識別する方法 (EC2 / RDS MySQL) https://qiita.com/hirooka622/items/d9ffb3aaf5fbba0a8a8d 1 2 3 4 5 6 7 8 9 mysql> show variables like "hostname"; +---------------+---------------+ | Variable_name | Value | +---------------+---------------+ | hostname | ip-172-23-1-5 | +---------------+---------------+ 1 row in set (0.00 sec)

DRF: DataFrame を使って CSV エクスポート CSV レンダラー 1 2 3 4 5 6 7 def dataframe_to_csv_stream(df, encoding=None, **kwargs): buffer = BytesIO() df.to_csv(buffer, **kwargs) buffer.seek(0) if encoding: return BytesIO(buffer.read().decode("utf8").encode(encoding)) return buffer 1 2 3 4 5 6 7 8 9 10 class DataFrameCSVRenderer(BaseRenderer): media_type = "text/csv" format = "csv" def render(self, data, media_type=None, renderer_context=None, writer_opts=None): renderer_context = renderer_context or {} if not isinstance(data, pd.DataFrame): data = pd.DataFrame(data) stream = dataframe_to_csv_stream(data, index=False) return stream.read().decode("utf-8-sig") DataFrame を返すリストシリアライザ 1 2 3 4 5 6 7 class DataFrameSerializer(serializers.ListSerializer): @property def data(self): ret = super().data ser = ret.serializer.child columns = dict((name, field.label) for name, field in ser.fields.items()) return pd.DataFrame(ret).rename(columns=columns) エクスポートするシリアライザを生成するメタクラス 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 class ExpporSerializerMetaclass(serializers.SerializerMetaclass): EXPORT_META = {} FIELD_CLASS_DEFAULTS = {} @classmethod def create_fields(cls): def _create_field(item): label, name, klass = item defaults = cls.FIELD_CLASS_DEFAULTS[klass] return (name, getattr(serializers, klass)(label=label, **defaults)) return dict(map(_create_field, cls.EXPORT_META)) @classmethod def annotate(cls, queryset): raise NotImplementedError() def __new__(cls, name, bases, attrs, **kwargs): attrs["Meta"].fields = [i[1] for i in cls.EXPORT_META] # CSVで返答するカラム attrs["Meta"].list_serializer_class = DataFrameSerializer # リストシリアライザ attrs.update(**cls.create_fields()) attrs["annotate"] = cls.annotate # クエリセットをアノテートさせるクラスメソッド return super().__new__(cls, name, bases, attrs, **kwargs) 実際のメタクラス 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 class IncomeExpporSerializerMetaclass(ExpporSerializerMetaclass): # (ラベル, フィールド名, フィールドクラス名) EXPORT_META = [ ("会社コード", "order__company_code", "CharField"), .... ] # フィールドクラスのデフォルト値 FIELD_CLASS_DEFAULTS = dict( CharField=dict(read_only=True), DecimalField=dict(max_digits=12, decimal_places=1, read_only=True), ) @classmethod def annotate(cls, queryset): """ クエリセットを適切にアノテートする""" def _ann(item): label, name, klass = item if name.startswith("order__"): # リレーション return (name, F(name)) if name.startswith("meme__"): # 固定値 return (name, Value("")) # その他はモデルフィールドなのでアノテートしない return (None, None) ann = dict(filter(lambda i: i[1] is not None, map(_ann, cls.EXPORT_META))) return queryset.annotate(**ann) 実際のシリアライザ 1 2 3 class IncomeExportSerializer(serializers.ModelSerializer, metaclass=IncomeExpporSerializerMetaclass): class Meta: model = models.Income API 本体(ビューセット) 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 class IncomeViewSet(VS.BaseModelViewSet): serializer_class = serializers.IncomeSerializer queryset = models.Income.objects.all() pagination_class = paginations.Pagination renderer_classes = tuple(api_settings.DEFAULT_RENDERER_CLASSES) + (DataFrameCSVRenderer,) def get_serializer_class(self): """ export アクションでシリアライザを切り替える """ res = { "export": serializers.IncomeExportSerializer, }.get(self.action, super().get_serializer_class()) return res def get_queryset(self): """ export アクションの場合、必要なアノテーションを行う """ qs = super().get_queryset() if self.action == "export": qs = serializers.IncomeExportSerializer.annotate(qs).filter(level=0) return qs return qs @decorators.action(methods=["get"], detail=False) def export(self, request, pk=None): ## リスト処理を行うと シリアライザとレンダラで勝手にCSVを返答する(Accept: text/csvの時) return self.list(request)

pandas : SettingWithCopyWarning: A value is trying to be set on a copy of a slice from a DataFrame. どちらでも Warning: 1 2 3 4 5 6 def _balance(df): if df["bookitem_kind"] == "1": return -1 * df["amount"] return df["amount"] df_epm_cost["amount2"] = df_epm_cost.apply(_balance, axis=1) 1 df_epm_cost["amount"].where(df_epm_cost["bookitem_kind"] == "0", -1 * df_epm_cost["amount"], inplace=True) ValueError: cannot reindex on an axis with duplicate labels 1 df_epm_cost.loc[df_epm_cost["bookitem_kind"] == "1", "amount2"] = df_epm_cost["amount"] * (-1) FutureWarning: reindexing with a non-unique Index is deprecated and will raise in a future version ValueError: cannot reindex on an axis with duplicate labels インデックスを削除するとうまく行く: ...

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

MySQL バルクインサート MySQL 複数データのバルク INSERT や CSV ファイルから高速インポートする方法 【MySQL】LOAD DATA INFILE するときのファイルの文字コード 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 @main.command() @click.argument("source_file") @click.option("--encoding", "-e", default="cp932") @click.pass_context def data_import(ctx, source_file, encoding): """ import TSV file """ opt = get_model()._meta fields = ",".join([f.name for f in opt.fields if f.name not in ["id"]]) SQL = f""" LOAD DATA LOCAL INFILE '{source_file}' INTO TABLE {opt.db_table} CHARACTER SET {encoding} FIELDS TERMINATED BY '\t' ({fields}) ; """ with connection.cursor() as cursor: cursor.execute(SQL) print(cursor.fetchall()) 重複時に UPDATE する(REPLACE): ...

pandas: Dataframe pandas で特定の文字列を含む行を抽出（完全一致、部分一致） 複数条件: &, | 否定: ~ (NOT) 型変換 Python: pandas でカラムの型を変換する 検索 選択肢: .isin(list) 1 2 result = df[df["code"].isin(["0", "1"])] # 0,1 のみ result = df[~(df["code"].isin(["0", "1"]))] # 0, 1以外 ユニーク unique: 1 items = df["code"].unique() 重複を除外(drop_duplicates): 1 result = df[["code", "name"]].drop_duplicates() 重複を抽出: 1 2 3 4 size = df.groupby(keys).size() res = size[ size > 1 ] if res.shape[0] > 0: print(res) 変換 dict 一覧: ...

ruff 新しい静的コード解析ツール「Ruff」をご紹介 Ruff の紹介 Python の Ruff (linter) でコード整形もできるようになりました Python の Linter Formatter は、もう Ruff 一択。最短 5 分でプロジェクトに導入 1 2 poetry run ruff format . poetry run ruff check . --fix

Pandas: ソート Find the unique values in a column and then sort them pandas unique values multiple columns NumPy 配列 ndarray を一次元化（平坦化）する ravel と flatten

EBSByteBalance% AWS: EBS 最適化 RDS インスタンスクラスを使用している場合は、CloudWatch のグラフを使用して IOPS またはスループットのスロットリングがないかを確認します。 バーストキャパシティを備えたインスタンスクラスの場合は、CloudWatch のグラフで EBSIOBalance% および EBSByteBalance% のメトリクスを表示します。 EBSIOBalance% または EBSByteBalance% の常に低い値は、インスタンスレベルで IOPS またはスループットのボトルネックが発生していることを示唆しています。 資料 Amazon RDS インスタンスの IOPS のボトルネックによって引き起こされた Amazon EBS ボリュームのレイテンシーをトラブルシューティングするにはどうすればよいですか?

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