問題タブ [snowflake-cloud-data-platform]

私の機能の一部にはSnowflakeでより高速なランタイム分析が必要なため、PostgresからSnowFlakeに移行することを検討しています。

Djago ORM クエリの代わりに生のクエリを作成する必要がある Snowflake 用の Python コネクタ API しか取得できませんでした。

とにかく、Django を使用して Snowflake に接続し、Snowflake で同じ Django ORM クエリを実行できますか。

この古い投稿を見まし た djangoでこの（スノーフレーク）データスキーマを照会する方法

しかし、Snowflake との django 接続をセットアップする方法が見つかりませんでした。

Apache 1.9.0 を使用しています。スノーフレーク フック プラグインを作成しました。$AIRFLOW_HOME/plugins ディレクトリにフックを配置しました。

snowflake_hook2.py

そこで、Airflows plugin_manager に print ステートメントをいくつか入れて、何が起こっているのかをより適切に処理できるようにしました。ウェブサーバーを再起動して airflow list_dags を実行すると、これらの行に「新しいモジュール名」が表示されました (エラーはありません)

これはドキュメントの内容と一致しているため、DAG でこれを使用しても問題ありません。

しかし、ウェブはこのエラーをスローします

だから問題は、私は何を間違っているのですか？または、バグを発見しましたか？

SnowflakeConnection クラスのラッパーとして機能する Python の SnowflakeApi クラスがあります。私のSnowflakeApiは

SnowflakeApi の名前空間は connection.snowflake_connection.SnowflakeApi です (つまり、connections というフォルダーに snowflake_connection.py があります)

pytest と unittest.mock を使用して、このクラスの単体テストを作成したいと考えています。問題は、MagicMock オブジェクトが返され、データベース呼び出しが行われないように、'connect' をモックしたいことです。これまでのところ、私は試しました：

1. monkeypatch.setattr(connections.snowflake_connection,"接続",return_value = "")
2. 元のクラスを変更して、スノーフレークをインポートするだけにしました。次に、モック オブジェクトを作成し、monkeypatch.setattr(snowflake_connection,"snowflake",my_mock_snowflake) を使用しました。それもうまくいきませんでした

要するに、他にもいくつか試してみましたが、何もうまくいきませんでした。私がやりたいのは、スノーフレーク接続を模擬することだけなので、実際のデータベース呼び出しは行われません。