feat:Product Serving 리뉴얼

.gitignore

@@ -2,4 +2,6 @@
 .vscode/
 .idea/
 .venv/
-__pycache__/
+__pycache__/
+.ipynb_checkpoints/
+mlruns/

01-batch-serving(airflow)/dags/05-python-operator-with-slack-noti.py

@@ -0,0 +1,39 @@
+from airflow import DAG
+from airflow.operators.python import PythonOperator
+from datetime import datetime, timedelta
+
+from airflow.exceptions import AirflowFailException
+from operators.slack_notifier import task_fail_slack_alert, task_succ_slack_alert
+
+
+# slack_notifier 에 선언한 webhook 전송 함수를 활용하여 slack 알림을 제공합니다
+default_args = {
+    'owner': 'kyle',
+    'depends_on_past': False,
+    'start_date': datetime(2022, 4, 20),
+    'retires': 1,
+    'retry_delay': timedelta(minutes=5),
+    'on_failure_callback': task_fail_slack_alert, # 실패 알림
+    # 'on_success_callback': task_succ_slack_alert, # 성공 알림 필요 시 추가
+}
+
+
+def _handle_job_error() -> None:
+    raise AirflowFailException("Raise Exception.")
+
+
+with DAG(
+        dag_id='python_dag_with_slack_webhook',
+        default_args=default_args,
+        schedule_interval='30 0 * * *',
+        tags=['my_dags']
+) as dag:
+    execution_date = "{{ ds }}"
+
+    send_slack_noti = PythonOperator(
+        task_id='raise_exception_and_send_slack_noti',
+        python_callable=_handle_job_error,
+        op_args=[execution_date]
+    )
+
+    send_slack_noti

01-batch-serving(airflow)/data/.gitignore

@@ -0,0 +1 @@
+

part4/02-airflow/docker-compose.yml → 01-batch-serving(airflow)/docker-compose.yml

@@ -8,7 +8,7 @@ services:
     image: postgres:13
     environment:
       - POSTGRES_USER=airflow
-      - POSTGRES_PASSWORD=1234
+      - POSTGRES_PASSWORD=!boostcamp-aitech!
     volumes:
       - ./data:/var/lib/postgresql/data
     restart: always
@@ -27,11 +27,11 @@ services:
         airflow db init && \
         airflow users create \
           --username admin \
-          --password 1234 \
-          --firstname heumsi \
-          --lastname jeon \
+          --password !boostcamp-aitech! \
+          --firstname kyle \
+          --lastname byeon \
           --role Admin \
-          --email heumsi@naver.com \
+          --email snugyun01@gmail.com \
         "
     restart: on-failure
     networks:
@@ -72,7 +72,7 @@ services:
       - airflow-init
     image: codercom/code-server:4.0.1
     environment:
-      - PASSWORD=1234
+      - PASSWORD=!boostcamp-aitech!
       - HOST=0.0.0.0
       - PORT=8888
     volumes:

part4/02-airflow/docker-readme.md → 01-batch-serving(airflow)/docker-readme.md

@@ -42,7 +42,7 @@ $ docker run \
   --network airflow \
   -v $(pwd)/data:/var/lib/postgresql/data \
   -e POSTGRES_USER=airflow \
-  -e POSTGRES_PASSWORD=1234 \
+  -e POSTGRES_PASSWORD=!boostcamp-aitech! \
   postgres:13
 ```
 
@@ -70,11 +70,11 @@ $ docker run \
     airflow db init && \
     airflow users create \
       --username admin \
-      --password 1234 \
-      --firstname heumsi \
-      --lastname jeon \
+      --password !boostcamp-aitech! \
+      --firstname seongyun \
+      --lastname byeon \
       --role Admin \
-      --email heumsi@naver.com \
+      --email snugyun01@gmail.com \
   "
 ```
 
@@ -141,7 +141,7 @@ $ docker run -it --name code-server \
     -d \
     -v "$(pwd)/dags:/home/coder/project" \
     -p 8888:8888 \
-    -e PASSWORD=1234 \
+    -e PASSWORD=!boostcamp-aitech! \
     -e HOST=0.0.0.0 \
     -e PORT=8888 \
     codercom/code-server:4.0.2

01-batch-serving(airflow)/operators/slack_notifier.py

@@ -0,0 +1,47 @@
+from airflow.providers.slack.operators.slack_webhook import SlackWebhookOperator
+
+# Slack Webhook 제공 Operator 를 먼저 정의합니다
+# 1. Connection ID (실습에서 Webserver 를 통해 생성한 값)
+SLACK_DAG_CONN_ID = "my_webhook"
+
+
+# 2. Webhook 함수 정의
+def send_message(slack_msg):
+    return SlackWebhookOperator(
+        task_id="slack_webhook",
+        slack_webhook_conn_id=SLACK_DAG_CONN_ID,
+        message=slack_msg,
+        username="sample-airflow",
+    )
+
+
+# 3. slack alert 함수 정의
+# 메시지에 Slack ID 추가해 tag 가능 (ex. <@U022T50D4F5>)
+def task_fail_slack_alert(context):
+    slack_msg = """
+            :red_circle: Task Failed.
+            *Task*: {task}  
+            *Dag*: `{dag}` 
+            *Execution Time*: {exec_date}
+            """.format(
+        task=context.get("task_instance").task_id,
+        dag=context.get("task_instance").dag_id,
+        exec_date=context.get("execution_date"),
+    )
+
+    alert = send_message(slack_msg)
+
+    return alert.execute(context=context)
+
+
+def task_succ_slack_alert(context):
+    slack_msg = f"""
+            :large_green_circle: Task SUCC.
+            *Task*: {context.get("task_instance").task_id}  
+            *Dag*: {context.get("task_instance").dag_id} 
+            *Execution Time*: {context.get("execution_date")}
+            """
+
+    alert = send_message(slack_msg)
+
+    return alert.execute(context=context)

01-batch-serving(airflow)/requirements.txt

@@ -0,0 +1 @@
+apache-airflow==2.7.3

part3/01-fastapi/examples/11_event_handler.py → 02-online-serving(fastapi)/examples/11_lifespan.py

@@ -1,31 +1,31 @@
+from contextlib import asynccontextmanager
+
 from fastapi import FastAPI
 import uvicorn
 
-app = FastAPI()
-
 items = {}
 
 
-@app.on_event("startup")
-def startup_event():
+@asynccontextmanager
+async def lifespan(app: FastAPI):
     print("Start Up Event")
     items["foo"] = {"name": "Fighters"}
     items["bar"] = {"name": "Tenders"}
 
+    yield
 
-@app.on_event("shutdown")
-def shutdown_event():
     print("Shutdown Event!")
     with open("log.txt", mode="a") as log:
         log.write("Application shutdown")
 
 
+app = FastAPI(lifespan=lifespan)
+
+
 @app.get("/items/{item_id}")
 def read_items(item_id: str):
     return items[item_id]
 
 
 if __name__ == '__main__':
-    uvicorn.run(app, host="0.0.0.0", port=8000)
-
-
+    uvicorn.run(app, host="0.0.0.0", port=8000)