Forem: Tanakrit Seangnet

🚀 ทำความเข้าใจ Row-Level Security (RLS) ใน PostgreSQL แบบง่ายที่สุด

Tanakrit Seangnet — Thu, 13 Nov 2025 03:27:17 +0000

Row-Level Security (RLS) เป็นหนึ่งในฟีเจอร์สำคัญของ PostgreSQL ที่ช่วยให้ระบบสามารถควบคุมการเข้าถึงข้อมูล ระดับแถว ได้อย่างปลอดภัย โดยเฉพาะระบบที่มีผู้ใช้หลายกลุ่ม เช่น โรงพยาบาล, องค์กร, หรือ Multi-tenant application

บทความนี้จะอธิบาย

RLS คืออะไร
สิทธิของ superuser / owner ส่งผลยังไง
ตัวอย่างผลลัพธ์ “ก่อน” และ “หลัง” ใช้ RLS

📌 Row-Level Security (RLS) คืออะไร?

RLS = การกรองข้อมูลระดับแถวโดยอัตโนมัติ ตามสิทธิของ user

PostgreSQL จะตรวจสอบทุกคำสั่ง SELECT, UPDATE, DELETE, INSERT

แล้วแสดงเฉพาะแถวที่ผ่าน policy ที่เรากำหนดไว้ เช่น:

เห็นเฉพาะข้อมูลของโรงพยาบาลตนเอง
เห็นเฉพาะข้อมูลที่ active
ห้ามแก้ไขข้อมูลที่ถูกซ่อน
ห้ามลบข้อมูลที่ไม่ใช่เจ้าของ

สามารถคิดง่ายๆ ว่า เป็นเหมือน WHERE อัตโนมัติที่ซ่อนอยู่ในระบบ

และ PostgreSQL จะใช้มันเองเสมอเมื่อมีการ Query

📌 สิทธิที่เกี่ยวข้องกับ RLS

✔ 1. Superuser

ไม่ถูกบังคับด้วย RLS
มองเห็นทุกเรคคอร์ดเหมือนปิด RLS อยู่
ไม่ต้องมี policy ก็เห็นทุกข้อมูล

superuser = bypass RLS

✔ 2. Owner ของ table

ค่าเริ่มต้น owner ก็ ไม่โดน RLS เช่นกัน

owner = select ได้ทุกแถว
owner = update/delete ได้ทุกแถว
แม้เปิด RLS แล้วก็ตาม

ถ้าอยากให้ owner ต้องทำตาม policy ต้องใช้:

ALTER TABLE your_table FORCE ROW LEVEL SECURITY;

หลังจากใช้ FORCE:

owner ต้องทำตาม policy เช่นเดียวกับ user คนอื่น
ยกเว้น superuser ที่ยัง bypass อยู่

✔ 3. User ปกติ (non-owner)

เป็นกลุ่มที่ RLS จะมีผลจริง

ถ้าเปิด RLS แต่ไม่มี policy → เห็น 0 rows
ถ้ามี policy → เห็นเฉพาะข้อมูลที่ตรง policy

📌 วิธีเปิด / ปิด RLS

เปิด RLS:

ALTER TABLE t_order ENABLE ROW LEVEL SECURITY;

ปิด RLS:

ALTER TABLE t_order DISABLE ROW LEVEL SECURITY;

📘 ตัวอย่างเข้าใจง่ายที่สุด

สมมติเรามี table:

CREATE TABLE t_order (
   id SERIAL PRIMARY KEY,
   hospital_code TEXT,
   patient_name TEXT
);

INSERT INTO t_order (hospital_code, patient_name) VALUES
('A', 'Alice'),
('A', 'Amy'),
('B', 'Bob'),
('C', 'Chris');

User ที่เข้าถึงระบบ:

userA → โรงพยาบาล A
userB → โรงพยาบาล B

🟦 ก่อนเปิด RLS (ไม่ใช้ RLS)

ผู้ใช้ทั้งสองเห็นเหมือนกันหมด

userA:

SELECT * FROM t_order;

ผลลัพธ์:

id	hospital_code	patient_name
1	A	Alice
2	A	Amy
3	B	Bob
4	C	Chris

userB:

ได้ผลเหมือนกันทุกอย่าง

➡ ทุกคนเห็นทุกข้อมูล

🟩 เปิดใช้งาน RLS

เปิด RLS:

ALTER TABLE t_order ENABLE ROW LEVEL SECURITY;

สร้าง policy จำกัดให้เห็นเฉพาะโรงพยาบาลของตน:

CREATE POLICY order_filter ON t_order
    FOR SELECT
    USING (hospital_code = current_setting('myapp.hospital_code'));

ก่อนสั่ง query ให้ตั้งค่าตาม user

🔍 ผลลัพธ์เมื่อเปิด RLS

👉 userA:

SET myapp.hospital_code = 'A';
SELECT * FROM t_order;

ผลลัพธ์:

id	hospital_code	patient_name
1	A	Alice
2	A	Amy

👉 userB:

SET myapp.hospital_code = 'B';
SELECT * FROM t_order;

ผลลัพธ์:

id	hospital_code	patient_name
3	B	Bob

🔥 สรุปผลลัพธ์แบบเทียบกันชัดๆ

❌ ไม่ใช้ RLS

User	ข้อมูลที่เห็น
userA	ทุกแถว
userB	ทุกแถว

✅ ใช้ RLS

User	ข้อมูลที่เห็น
userA	เฉพาะโรงพยาบาล A
userB	เฉพาะโรงพยาบาล B

⚠ หมายเหตุสำคัญ

หากเปิด RLS แต่ ไม่มี policy

ผู้ใช้ปกติจะเหมือนว่า table ว่างทันที:

SELECT * FROM t_order;

ผล:

0 rows

แต่ superuser และ owner ยังเห็นได้ (ถ้าไม่ FORCE)

📌 สรุปทั้งหมด

RLS คือฟีเจอร์ควบคุมการเห็นข้อมูลระดับแถว
Superuser และ Owner (โดย default) จะ bypass RLS
User ปกติจะถูกบังคับใช้ RLS
เปิด RLS แล้ว ต้องมี policy ไม่งั้นเห็น 0 rows
ใช้ RLS เพื่อทำระบบ multi-tenant หรือแบ่งสิทธิข้อมูลได้อย่างปลอดภัยมาก

ทดสอบ API อัตโนมัติด้วย Postman และ Newman

Tanakrit Seangnet — Tue, 14 Jan 2025 08:06:14 +0000

การทดสอบ API เป็นขั้นตอนที่สำคัญในการพัฒนาและบำรุงรักษาแอปพลิเคชัน ซึ่ง Postman เป็นเครื่องมือที่มีชื่อเสียงสำหรับทำการทดสอบ API อย่างมีประสิทธิภาพ ในบทความนี้เราจะพาลงลึกไปในกระบวนการทดสอบอัตโนมัติ API โดยใช้ Postman และ Newman ซึ่งเป็นเครื่องมือที่ช่วยในการรันคำสั่งทดสอบของ Postman ได้อย่างสะดวกและรวดเร็ว

Postman คืออะไร ?

Postman เป็นเครื่องมือที่มีหน้าตาใช้งานง่ายและมีความสามารถมากมาย ทำให้นักพัฒนาสามารถทดสอบ API ได้อย่างรวดเร็วและมีประสิทธิภาพ ติดตั้ง Postman และเริ่มต้นการทดสอบ API โดยการส่งคำขอ GET, POST, PUT, และ DELETE ไปยัง API ต่าง ๆ ใน Postman, สามารถเขียนสคริปต์ทดสอบเพื่อตรวจสอบผลลัพธ์ที่ได้จาก API ได้ นักพัฒนาสามารถเขียนสคริปต์ที่ทำการตรวจสอบค่าที่คืนมาจาก API ว่าตรงตามคาดหวังหรือไม่

Newman คืออะไร ?

Newman เป็นเครื่องมือที่ทำให้เราสามารถรันคำสั่งทดสอบของ Postman อัตโนมัติได้ นักพัฒนาสามารถใช้คำสั่งในโปรแกรมหรือ CI/CD pipeline เพื่อรันทดสอบอัตโนมัติและตรวจสอบความถูกต้องของ API

การใช้งาน Postman

การสร้าง Request:

เปิด Postman และคลิกที่ "New" เพื่อสร้าง Request ใหม่.
ใส่ URL ของ API ที่ต้องการทดสอบ.

การกำหนด Method:

เลือก HTTP Method เช่น GET, POST, PUT, หรือ DELETE.

การเพิ่ม Headers:

ใส่ Headers ที่จำเป็นต้องใช้ เช่น Content-Type, Authorization.

การส่งข้อมูล:

ถ้าต้องการส่งข้อมูลเพิ่มเติม (เช่น JSON หรือ form data) ให้ใส่ข้อมูลในส่วนของ Body.

การทดสอบ Request:

กดปุ่ม "Send" เพื่อทดสอบ Request และดูผลลัพธ์.

การจัดการ Environment:

ใช้ Environment เพื่อจัดการตัวแปรและค่าที่ใช้ซ้ำในหลาย Request.

การใช้ Collections:

สร้าง Collections เพื่อจัดการกลุ่มของ Requests ที่เกี่ยวข้อง.

การทดสอบ Script:

สามารถเขียน JavaScript ในส่วนของ Pre-request Script หรือ Tests เพื่อปรับแต่งหรือตรวจสอบผลลัพธ์.

การจัดการการตอบรับ (Response):

ตรวจสอบ Response ที่ได้รับ เพื่อให้แน่ใจว่ามีข้อมูลที่ถูกต้อง.

การเขียน Test script

ตัวอย่างข้อมูลที่ใช้สำหรับเขียน Test script

Method GET: https://jsonplaceholder.typicode.com/posts
Method POST: https://jsonplaceholder.typicode.com/posts
Request Body : {
"title": "foo",
"body": "bar",
"userId": 1
}
Method PUT: https://jsonplaceholder.typicode.com/posts/:id
Request Param: id
Method DELETE: https://jsonplaceholder.typicode.com/posts/:id
Request Param: id

Method GET

ในตัวอย่างนี้เราต้องการเขียน Test script สำหรับตรวจสอบข้อมูลดังนี้

ตรวจสอบ status code ว่าตอบกลับเป็น 200 ไหม
ตรวจสอบ response body ว่ามีคำว่า aut อยู่ไหม
ตรวจสอบ header ว่ามี key Content-Type อยู่ไหม
ในกรณีตรวจสอบไม่ผ่าน

// ตรวจสอบ status code 200
pm.test("Status code is 200", function () {
    pm.response.to.have.status(200);
});

// ตรวจสอบ Response body ว่ามีคำว่า aut อยู่ไหม
pm.test("Response body success", function () {
    pm.expect(pm.response.text()).to.include("aut");
});

// ตรวจสอบ Response header มี Content-Type อยู่ไหม
pm.test("Content-Type is present", function () {
    pm.response.to.have.header("Content-Type");
});

// ตัวอย่างตรวจสอบ FAIL
pm.test("Status code is 201", function () {
    pm.response.to.have.status(201);
});

Method POST

ในตัวอย่างนี้เราต้องการเขียน Test script สำหรับตรวจสอบข้อมูลดังนี้

ตรวจสอบ status code ว่าตอบกลับเป็น 201 ไหม
ตรวจสอบ Response body โดย id = 101, body = bar

// ตรวจสอบ status code
pm.test("Successful POST request", function () {
    pm.expect(pm.response.code).to.be.oneOf([201, 202]);
});

// ตรวจสอบ Response body โดย id = 101, body = bar
pm.test("Response body success", function () {
    var jsonData = pm.response.json();
    pm.expect(jsonData.id).to.eql(101);
    pm.expect(jsonData.body).to.eql("bar");
});

การใช้งาน newman

ติดตั้ง Node.js
เนื่องจาก Newman ทำงานผ่าน Node.js ให้ดาวน์โหลดและติดตั้ง Node.js จาก Node.js Official Website
(แนะนำให้เลือก LTS Version เพื่อความเสถียร)

// ติดตั้ง Newman
npm install -g newman

// ติดตั้งเครื่องมือสำหรับออกรายงาย
npm install newman-reporter-html

// ตรวจสอบการติดตั้ง Newman
newman -v

การใช้งาน Newman คู่กับ Postman

สร้างและ Export Collection จาก Postman

เปิด Postman และเลือก Collection ที่ต้องการทดสอบ
คลิกขวาที่ Collection และเลือก Export
เลือกรูปแบบเป็น v2.1 (recommended) จากนั้นบันทึกไฟล์ .json ลงในเครื่อง

รัน Collection ด้วย Newman

ใช้คำสั่งด้านล่างเพื่อรัน Collection ที่ export มาจาก Postman:

newman run <path_to_collection_file.json>

เพิ่ม Environment File (ถ้ามี)

หาก Collection ต้องการ Environment Variables:

Export Environment จาก Postman ในรูปแบบ .json
ใช้คำสั่งนี้เพื่อรัน Collection พร้อม Environment:

newman run <path_to_collection_file.json> -e <path_to_environment_file.json>

การสร้างรายงานผลการทดสอบ

Newman สามารถสร้างรายงานการทดสอบในรูปแบบต่างๆ ได้ เช่น HTML หรือ JSON
ตัวอย่างการสร้างรายงาน HTML:

newman run <path_to_collection_file.json> -r html

เคล็ดลับเพิ่มเติม

ใช้ Postman Monitor ในการรัน Collection แบบอัตโนมัติ หากต้องการใช้งานบนคลาวด์
ใช้ Plugin อย่าง newman-reporter-htmlextra เพื่อสร้างรายงาน HTML ที่มีข้อมูลมากขึ้น

npm install -g newman-reporter-htmlextra
newman run MyCollection.json -r htmlextra

คู่มือการใช้งาน Airflow

Tanakrit Seangnet — Wed, 03 Mar 2021 02:34:30 +0000

สิ่งที่ต้องมี

Docker

เปิดใช้งาน Airflow

ขั้นตอนที่ 1. แตกไฟล์โปรเจคที่ได้รับ

./dags - สามารถใส่ไฟล์ DAG ของคุณได้ที่นี่
./logs - บันทึกจาก execution และ scheduler
./plugins - สามารถใส่ปลั๊กอินที่กำหนดเองได้ที่นี่
./worker - ไฟล์ airflow.cfg

ขั้นตอนที่ 2. เรียกใช้งาน Terminal cd มาที่โฟลเดอร์

ใช้คำสั่งและรอจนกว่าขึ้นแบบในรูป

docker-compose up airflow-init

ขั้นตอนที่ 3. ใช้คำสั่งเพื่อเริ่มต้นใช้งาน

ก่อนเริ่มใช้งานควรปรับแต่ง config ให้เรียบร้อย

docker-compose up

ล็อกอินเพื่อเข้าใช้งาน web ui ได้ที่ http://localhost:8080.
ชื่อบัญชีผู้ใช้ที่ถูกสร้างขึ้นคือ airflow และรหัสผ่าน airflow สำหรับ admin

การปรับแต่ง Config

การปรับแต่งจะทำในไฟล์ ./worker/airflow.cfg

Timezone

#มี default เป็น UTC
[core]
default_timezone = Asia/Bangkok

[webserver]
default_ui_timezone = Asia/Bangkok

SMTP Server

ใช้ SMTP Server Gmail สำหรับส่งอีเมล

[smtp]
smtp_host = smtp.gmail.com
smtp_starttls = True
smtp_ssl = False
# Example: smtp_user = airflow
smtp_user = your@gmail.com
# Example: smtp_password = airflow
smtp_password = your_password_gmail
smtp_port = 25 หรือ 587
smtp_mail_from = your@gmail.com
smtp_timeout = 30
smtp_retry_limit = 5

การใช้งาน Web UI

การสร้าง Connection ID

ระบุข้อมูลเพื่อเชื่อมต่อกับฐานข้อมูล conn_id จะต้องตรงกับ CONN_ID ในไฟล์ python

การสร้าง Users

กดปุ่ม + เพื่อเพิ่ม User

การสร้าง Roles

กดปุ่ม + เพื่อเพิ่ม Roles

การเปิดใช้งาน Active DAG

การ Trigger DAG

Line Notify in Airflow

Tanakrit Seangnet — Mon, 01 Mar 2021 10:20:29 +0000

workflow

วันนี้จะมาสร้าง workflow ง่ายๆ ใช้งานกัน

ดึงข้อมูล Covid-19 จาก API ของกรมควบคุมโรค
ส่งแจ้งเตือนข้อมูลเข้า Line notify ทุกๆ 8 โมงเช้า

ถ้ายังไม่รู้จักการใช้ Airflow ให้เข้าไปอ่านได้ที่นี้

เริ่มต้น

ต้องเป็นเพื่อนกับ Line Notify ก่อน ด้วยการค้นหาเพื่อนชื่อ Line Notify แล้วแอดเป็นเพื่อน

หลังจากนั้น ให้ไปที่ https://notify-bot.line.me/th/ แล้วเข้าสู่ระบบ

Login บัญชี LINE ด้วยอีเมล และ รหัสผ่าน

หลังจาก Login สำเร็จ ให้เลือกเมนูด้านขวาบนตรงชื่อ account ของเรา และเลือกที่หน้าของฉัน

เลื่อนลงมาด้านล่างให้กดปุ่ม “Generate token”

จากนั้นให้ใส่

ชื่อของ Token (จะแสดงตอนแจ้งเตือนข้อความ)
เลือกห้องแชทที่ต้องการส่งข้อความแจ้งเตือน จากนั้นกดปุ่มออก Token เพื่อรับ Token key

ให้คัดลอก Token key ไว้ อย่าเพิ่งปิด pop up ไม่อย่างนั้นต้องออก Token ใหม่

ในขณะเดียวกัน line notify ก็จะแจ้งเตือนเรา

หลังจากเราเตรียม Token เสร็จแล้วก็เริ่มเขียนโค้ดได้เลย

ดึงข้อมูลจาก API

import json
import requests

def get_covid19_report_today():
    url = 'https://covid19.th-stat.com/api/open/today'
    response = requests.get(url)
    data = response.json()

    with open('data.json', 'w') as f:
        json.dump(data, f)

    return data

ได้ข้อมูลจาก API แล้วก็ส่งข้อมูลเข้า Line notify ผ่าน API กันเลย
เอา Token มาใส่แทน your_token ได้เลย

def send_line_notify():
    url = 'https://notify-api.line.me/api/notify'
    token = 'your_token'
    headers = {
        'content-type':
        'application/x-www-form-urlencoded',
        'Authorization': 'Bearer '+token
    }

    with open('data.json') as f:
        data = json.load(f)

    msg = "Covid-19 report today \n"
    for (k,v) in data.items():
        msg += str(k) + ":" + str(v) + "\n"

    r = requests.post(url, headers=headers, data={'message': msg})
    print(r.text)

อย่าลืมสร้าง DAG object ด้วยนะ

schedule_interval กำหนดเป็น 0 8 * * * เราต้องการให้ทำงานทุกๆ 8 โมงของทุกวัน

from datetime import datetime
from airflow import DAG
from airflow.operators.python_operator import PythonOperator

default_args = {
    'owner': 'airflow',
    'start_date': datetime(2021, 2, 28, 0, 0, 0),
}
with DAG('line-notify',
         schedule_interval='0 8 * * *',
         default_args=default_args,
         description='A simple data pipeline for line-notify',
         catchup=False) as dag:

    t1 = PythonOperator(
        task_id='get_covid19_report_today',
        python_callable=get_covid19_report_today
    )

    t2 = PythonOperator(
        task_id='send_line_notify',
        python_callable=send_line_notify
    )

    t1 >> t2

ภายในไฟล์ python ที่เราจะได้

import requests
import json
from datetime import datetime

from airflow import DAG
from airflow.operators.python_operator import PythonOperator


def get_covid19_report_today():
    url = 'https://covid19.th-stat.com/api/open/today'
    response = requests.get(url)
    data = response.json()

    with open('data.json', 'w') as f:
        json.dump(data, f)

    return data


def send_line_notify():
    url = 'https://notify-api.line.me/api/notify'
    token = 'your_token'
    headers = {
        'content-type':
        'application/x-www-form-urlencoded',
        'Authorization': 'Bearer '+token
    }

    with open('data.json') as f:
        data = json.load(f)

    msg = "Covid-19 report today \n"
    for (k,v) in data.items():
        msg += str(k) + ":" + str(v) + "\n"

    r = requests.post(url, headers=headers, data={'message': msg})
    print(r.text)


default_args = {
    'owner': 'airflow',
    'start_date': datetime(2021, 2, 28, 0, 0, 0),
}
with DAG('line-notify',
         schedule_interval='0 8 * * *',
         default_args=default_args,
         description='A simple data pipeline for line-notify',
         catchup=False) as dag:

    t1 = PythonOperator(
        task_id='get_covid19_report_today',
        python_callable=get_covid19_report_today
    )

    t2 = PythonOperator(
        task_id='send_line_notify',
        python_callable=send_line_notify
    )

    t1 >> t2

แจ้งเตือนที่เราจะได้

Apache Airflow:Trigger DAG with REST API

Tanakrit Seangnet — Thu, 25 Feb 2021 10:13:02 +0000

Airflow ได้ยกเลิกการใช้งาน Experimental Rest API ไปตั้งแต่ v2.0.0 ด้วยเหตุผลว่า Experimental ไม่มีการควบคุมการเข้าถึง จึงได้เกิด Airflow API (Stable) (1.0.0) ขึ้นมา ดูเพิ่มเติม

มาทดลองใช้กัน

สิ่งที่ต้องมี

Postman

เปิดใช้งาน Authenticate

แก้ไข auth_backend ใน airflow.cfg

[api]
auth_backend = airflow.api.auth.backend.basic_auth

เราจำเป็นจะต้องเปิดการใช้ Authen ไม่งั้น API v1 จะปฏิเสธการร้องขอ (PermissionDenied) อย่าลืม restart ด้วย

ระบุ URL ลงใน Postman

ในกรณีที่ URL ของเราเป็น Localhost ในการ Trigger จะใช้ HTTP Methods:POST

http://localhost:8080/api/v1/dags/{DAG_ID}/dagRuns

กำหนด Authorization

ระบุ Type เป็น Basic Auth และ Username/Password ต้องถูกสร้างอยู่ใน airflow

กำหนด Headers

KEY: Content-Type VALUE: application/json

กำหนด Body

ระบุเป็น raw ประเภท JSON ใส่ค่า {}

Response

ถ้าไม่มีการทำงานที่ผิดพลาดได้จะรับ Response

{
    "conf": {},
    "dag_id": "covid19_daily",
    "dag_run_id": "manual__2021-02-25T10:04:41.703172+00:00",
    "end_date": null,
    "execution_date": "2021-02-25T10:04:41.703172+00:00",
    "external_trigger": true,
    "start_date": "2021-02-25T10:04:41.708731+00:00",
    "state": "running"
}

เรากลับไปเช็คที่หน้า Web UI ก็จะพบว่า DAG ของเรารันแล้ว เป็นการทำงานแบบ manual

ภายใน web ของ Airflow ก็จะมีคำอธิบาย วิธีการใช้งาน api ต่างๆให้ดู

ดูการใช้งาน Airflow REST API เพิ่มเติมได้ที่

https://airflow.apache.org/docs/apache-airflow/2.0.1/stable-rest-api-ref.html#section/Overview

การใช้งาน Apache Airflow บน Docker เบื้องต้น

Tanakrit Seangnet — Mon, 22 Feb 2021 07:55:40 +0000

Apache Airflow

Apache Airflow เป็น Open Source ที่เข้ามาจัดการ Task งานต่างๆ โดยต้องเขียนเป็น Python Code โดยแต่ละ Task สามารถดู Workflow การทำงานได้อย่างละเอียด

เราสามารถตั้งเวลาการทำงานคล้าย Crontab(Cron Jobs) บน Linux ได้ เช่น เมื่อรัน Job A ตอนเวลา 01.00 น.ทุกวัน รัน Job B ตอน 08.00 น. ในวันอาทิตย์ โดย Job ใน Airflow จะเรียกว่า DAGs (Directed Acyclic Graphs) มี Web UI ให้เรา Monitor Task Failured ที่เกิดขึ้นจากการทำงานในแต่ละ Task เราสามารถตั้งค่า Alert หากมี Job Failures เกิดขึ้น

Web UI

Web UI ของ Apache Airflow ที่ช่วยให้จัดการ DAG ต่างๆ

เริ่มการติดตั้ง

Step 1: git clone



git clone https://gitlab.com/tanakrit1/apache-airflow-with-docker-compose.git

config ต่างๆที่อยู่ภายใน docker-compose.yml

อ่าน docker-compose เพิ่มเติม

airflow-scheduler - The scheduler monitors all tasks and DAGs, then triggers the task instances once their dependencies are complete.
airflow-webserver - The webserver available at http://localhost:8080.
airflow-worker - The worker that executes the tasks given by the scheduler.
airflow-init - The initialization service.
flower - The flower app for monitoring the environment. It is available at http://localhost:8080.
postgres - The database.
redis - The redis - broker that forwards messages from scheduler to worker.

โฟลเดอร์จะถูกเชื่อมต่อระหว่าง host กับ container

./dags - you can put your DAG files here.
./logs - contains logs from task execution and scheduler.
./plugins - you can put your custom plugins here.

Step 2: ก่อนเริ่ม Start Airflow เป็นครั้งแรก

เราต้องเรียกใช้การสร้างฐานข้อมูลและสร้างบัญชีผู้ใช้แรก ในการดำเนินการให้เรียกใช้



docker-compose up airflow-init

หลังจากการเริ่มต้นเสร็จสมบูรณ์จะเห็นข้อความดังต่อไปนี้

ชื่อบัญชีผู้ใช้ที่ถูกสร้างขึ้นคือ airflow และรหัสผ่าน airflow

Step 3: เริ่มต้นการใช้งาน Airflow

ตอนนี้เราสามารถ Start Service ทั้งหมดด้วยคำสั่ง



docker-compose up

เราสามารถล็อกอินเข้าใช้งาน web ui ของ airflow ได้ที่ http://localhost:8080.

มาลองใช้งานกัน

Workflow ที่เราจะเขียนกันคือ

ดึงค่า covid-19 ประจำวันจาก API ของกรมควบคุมโรค
insert ข้อมูลลง postgres database
ส่งอีเมลเตือนการทำงาน (ใช้ Gmail SMTP Server)



CREATE SEQUENCE id;
CREATE TABLE public.daily_covid19_reports (
    id int4 NOT NULL DEFAULT nextval('id'::regclass),
    confirmed int4 NULL,
    recovered int4 NULL,
    hospitalized int4 NULL,
    deaths int4 NULL,
    new_confirmed int4 NULL,
    new_recovered int4 NULL,
    new_hospitalized int4 NULL,
    new_deaths int4 NULL,
    update_date timestamp NULL,
    "source" varchar(100) NULL,
    dev_by varchar(100) NULL,
    server_by varchar(100) NULL,
    CONSTRAINT daily_covid19_reports_pkey PRIMARY KEY (id)
);

โค้ดทั้งหมดอยู่ใน ./dags/covid-daily.py แล้ว

Step 4: สร้าง DAG object ด้วย python เอาไว้ใน ./dags



from airflow import DAG

default_args = {
    'owner': 'airflow',
    'start_date': datetime(2021, 2, 17),
    'schedule_interval': None,
}
with DAG('covid19_daily',
         schedule_interval='@daily',
         default_args=default_args,
         description='A simple data pipeline for COVID-19 report',
         catchup=False) as dag:

schedule_interval คือการสั่งรันเป็นเวลา สามารถใช้เครื่องมือกำหนดเวลา ได้ที่นี้

Step 5: ดึงข้อมูลจาก API



import json
import requests

def get_covid19_report_today():
    url = 'https://covid19.th-stat.com/api/open/today'
    response = requests.get(url)
    data = response.json()

    with open('data.json', 'w') as f:
        json.dump(data, f)

    return data

Step 6: insert ข้อมูลลง postgres database

สร้าง Connection ด้วย Web UI ของ airflow ไปที่ Admin > Connections
ระบุ conn id:airflow_db

ใช้ PostgresHook ในการเชื่อมต่อกับฐานข้อมูล



from airflow.hooks.postgres_hook import PostgresHook
from airflow.utils.dates import days_ago
from datetime import datetime

def insert_data():
    with open('data.json') as f:
        data = json.load(f)

    print(data) 

    mysql_hook = PostgresHook(postgres_conn_id='airflow_db')

    insert = """
        INSERT INTO public.daily_covid19_reports
            (confirmed, recovered, hospitalized, deaths, new_confirmed, new_recovered, new_hospitalized, new_deaths, update_date, "source", dev_by, server_by)
        VALUES(%s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s);
    """

    mysql_hook.run(insert, parameters=(data['Confirmed'],
                                       data['Recovered'],
                                       data['Hospitalized'],
                                       data['Deaths'],
                                       data['NewConfirmed'],
                                       data['NewRecovered'],
                                       data['NewHospitalized'],
                                       data['NewDeaths'],
                                       datetime.strptime(data['UpdateDate'], '%d/%m/%Y %H:%M'),
                                       data['Source'],
                                       data['DevBy'],
                                       data['SeverBy']))

Step 7: สร้าง Task ด้วย Operator และเชื่อมต่อ Task

ใช้ PythonOperator เพื่อไปใช้ function ที่เขียนขึ้น



from airflow.operators.python_operator import PythonOperator

    t1 = PythonOperator(
        task_id='get_covid19_report_today',
        python_callable=get_covid19_report_today
    )

    t2 = PythonOperator(
        task_id='insert_data',
        python_callable=insert_data
    )

    t1 >> t2

Step 8: สร้าง Task สำหรับส่งอีเมลด้วย EmailOperator

Setting up SMTP Server สำหรับส่งเมลด้วย Gmail แก้ไขภายในไฟล์ airflow.cfg โปรเจคนี้ได้เชื่อมต่อไฟล์นี้ระหว่าง host กับ container ให้แล้วอยู่ใน .\worker แก้ไขแล้วให้ restart docker-compose



[smtp]
smtp_host = smtp.gmail.com
smtp_starttls = True
smtp_ssl = False
smtp_user = YOUR_EMAIL_ADDRESS
smtp_password = 16_DIGIT_APP_PASSWORD
smtp_port = 587
smtp_mail_from = YOUR_EMAIL_ADDRESS

เพิ่ม Task ภายในไฟล์ Python



from airflow.operators.email_operator import EmailOperator

    t3 = EmailOperator(
        task_id='send_email',
        to=['your@gmail.com'],
        subject='Your covid19 report today is ready',
        html_content='Please check your dashboard. :)'
    )

    t1 >> t2 >> t3

ดูเพิ่มเติมได้ที่