Python을 사용하여 Snowflake 경고 관리하기¶

전제 조건¶

이 항목의 예제에서는 Snowflake와 연결하고 Snowflake Python APIs 을 사용할 수 있는 Root 오브젝트를 생성하는 코드를 추가했다고 가정합니다.

예를 들어, 다음 코드는 구성 파일에 정의된 연결 매개 변수를 사용하여 Snowflake에 대한 연결을 생성합니다.

from snowflake.core import Root
from snowflake.snowpark import Session

session = Session.builder.config("connection_name", "myconnection").create()
root = Root(session)

해당 코드에서는 결과 Session 오브젝트를 사용하여 API의 유형과 메서드를 사용하기 위해 Root 오브젝트를 생성합니다. 자세한 내용은 Snowflake Python APIs 을 사용하여 Snowflake에 연결 섹션을 참조하십시오.

경고 만들기¶

경고를 생성하려면 먼저 Alert 오브젝트를 생성한 다음 API Root 오브젝트에서 AlertCollection 오브젝트를 생성합니다. AlertCollection.create 를 사용하여, Snowflake에 새로운 경고를 추가합니다.

다음 예제의 코드는 계정에서 이름이 my_alert 인 경고를 나타내는 Alert 오브젝트를 생성합니다.

from snowflake.core.alert import Alert, MinutesSchedule

root.alerts.create(Alert(
    name="my_alert",
    condition="SELECT 1",
    action="SELECT 2",
    schedule=MinutesSchedule(minutes=1),
    comment="test comment"
))

이 코드는 AlertCollection 변수 alerts 를 생성하고 AlertCollection.create 를 사용하여 Snowflake에 새로운 경고를 생성합니다.

경고 세부 정보 가져오기¶

Alert 오브젝트를 반환하는 AlertResource.fetch 메서드를 호출하여 경고에 대한 정보를 얻을 수 있습니다.

다음 예제의 코드는 이름이 my_alert 인 경고에 대한 정보를 가져옵니다.

my_alert = root.alerts["my_alert"].fetch()
print(my_alert.to_dict())

경고 나열하기¶

Alert 오브젝트의 PagedIter 반복기를 반환하는 AlertCollection.iter 메서드를 사용하여 경고를 나열할 수 있습니다.

다음 예제의 코드는 이름이 my 로 시작하는 경고를 나열한 다음 각 경고의 이름을 인쇄합니다. 이 예제에서는 선택적 매개 변수 show_limit=5 를 설정하여 결과 수를 5 로 제한합니다.

alerts_iter = root.alerts.iter(like="my%", show_limit=5)
for alert_obj in alerts_iter:
  print(alert_obj.name)

경고 작업 수행하기¶

AlertResource 오브젝트를 사용하여 경고 실행 및 삭제와 같은 일반적인 경고 작업을 수행할 수 있습니다.

경고 리소스로 수행할 수 있는 일부 작업을 보여주기 위해, 다음 예제의 코드에서는 다음을 수행합니다.

1. my_alert 경고 리소스 오브젝트를 가져옵니다.

2. 경고를 실행합니다.

3. 경고를 삭제합니다.

my_alert_res = root.alerts["my_alert"]

my_alert_res.execute()
my_alert_res.drop()