alembic で modelクラスを全てまるっとインポートする

はじめに alembic 自動生成する際に、指定ディレクトリ配下のクラスを全てまるっとインポートする方法 問題 疑問 自動生成したい対象のModelクラスをインポートしないと生成してくれない まとめ 結果 解決方法 以下を env.py 先頭に追記して、 app/models 配下の全てのクラスをインポートする # Import all models in app.models model_path = os.path.dirname(os.path.join(os.path.dirname(os.path.abspath(__file__)), '../app/models/')) for py in [f[:-3] for f in os.listdir(model_path) if f.endswith('.py') and (f != '__init__.py' and f != 'base_model.py')]: mod = __import__('.'.join(['app.modesl', py])) classes = [getattr(mod, x) for x in dir(mod) if isinstance(getattr(mod, x), type)] for cls in classes: setattr(sys.modules[__name__], cls.__name__, cls) Refs python - Import all classes in directory?...

2024-07-04 ·  2024-08-02 · 1 分 · 76 文字

Alembic revision --autogenerate で変更がなくても空のrevisionファイルができるのを避ける

はじめに Alembic revision –autogenerate で変更がなくても空のrevisionファイルができるのを避ける python 3.11, alembic 1.7.7, sqlalchemy 2.0.30, sqlmodel 0.0.18 問題 疑問 変更がなくても空のrevisionファイルが生成されてしまう まとめ 結果 解決方法 以下のような process_revision_directives を設定することで、差分がない場合は空のrevisionファイルを生成しない def run_migrations_online(): : def process_revision_directives(context, revision, directives): if config.cmd_opts.autogenerate: script = directives[0] if script.upgrade_ops.is_empty(): directives[:] = [] print('No changes in schema detected.') : with connectable.connect() as connection: context.configure( connection=connection, target_metadata=target_metadata, include_schemas=True, dialect_opts={"paramstyle": "named"}, include_name=include_name, process_revision_directives=process_revision_directives, ) Refs python - How to prevent alembic revision –autogenerate from making revision file if it has not detected any changes?...

2024-07-04 ·  2024-08-02 · 1 分 · 81 文字

Alembic revision --autogenerate で差分のマイグレーションファイルができない

はじめに Alembic revision –autogenerate で差分のマイグレーションファイルができない python 3.11, alembic 1.7.7, sqlalchemy 2.0.30, sqlmodel 0.0.18 問題 毎回テーブル作成のマイグレーションファイルが生成されてしまう 解決方法 以下のような include_name を設定することで、デフォルトスキーマに限定する connection は schema 指定でアクセスしている為、type_=schema で name=None で検出される # exclude system table def include_name(name, type_, parent_names): if type_ == "schema": return name in [None] else: return True def run_migrations_online(): : with connectable.connect() as connection: context.configure( connection=connection, target_metadata=target_metadata, dialect_opts={"paramstyle": "named"}, include_schemas=True, include_name=include_name, ) Refs python - Alembic revision –autogenerate always detects a new table, instead of detecting new columns - Stack Overflow

2024-07-04 ·  2024-08-02 · 1 分 · 79 文字

SQLAlchemy で Flake8 error: E712 comparison to True should be 'if cond is True:' or 'if cond:' の lint エラーに対応する

発生したのは、以下のような sqlalchemy での条件文が flake8 の lint エラーになったので、対応したかった。 Flake8 error: E712 comparison to True should be 'if cond is True:' or 'if cond:' 実装していた処理は以下 ExampleModel.sa.delete_flag == True これが lint エラーになるからって、以下にすると、解消するけど、うまく動かない(条件判定がおかしくなる) 落とし穴!!! ExampleModel.sa.delete_flag is True # または ExampleModel.sa.delete_flag # 上記や、上記の否定など not ExampleModel.sa.delete_flag 結論としては、sqlalchemy.sql.expression.true, sqlalchemy.sql.expression.false あるからそれ使う # import from sqlalchemy import false, true # ussage ExampleModel.sa.delete_flag == true() いやいやいや… Refs python - Flake8 error: E712 comparison to True should be ‘if cond is True:’ or ‘if cond:’ - Stack Overflow python - SQLAlchemy Core select where condition contains boolean expression is False - Stack Overflow Column Elements and Expressions — SQLAlchemy 2....

2023-08-16 ·  2023-09-22 · 1 分 · 100 文字

Ubuntu 14.04 Python でWebsocket実装

Ubuntu14.04 Python 2.7.6 でWebsocket実装 モジュールのインストール sudo apt-get install libevent-dev python-dev python-setuptools sudo easy_install gevent sudo easy_install gevent-websocket サーバスクリプト作成 vi chat.py #! /usr/bin/python # -*- coding: utf-8 -*- import os import random from geventwebsocket.handler import WebSocketHandler from gevent import pywsgi, sleep ws_list = set() def chat_handle(environ, start_response): global cnt ws = environ['wsgi.websocket'] ws_list.add(ws)a print 'enter!', len(ws_list) while 1: msg = ws.receive() if msg is None: break remove = set() for s in ws_list: try: s....

2014-09-10 ·  2014-09-10 · 1 分 · 187 文字