From bcaa252b51ee1cba255d76ef162faa25fdd0895c Mon Sep 17 00:00:00 2001
From: trungtran <baotrung06092003@gmail.com>
Date: Fri, 6 Dec 2024 23:23:38 +0700
Subject: [PATCH] split product data from csv

---
 .DS_Store                                   | Bin 6148 -> 8196 bytes
 Dockerfile                                  |   1 +
 dags/init-db.py                             |   8 ++++-
 data/.DS_Store                              | Bin 0 -> 6148 bytes
 docker-compose.yaml                         |   3 ++
 plugins/models/models.py                    |   5 +--
 plugins/transform/transform_product_data.py |  33 ++++++++++++++++++++
 requirements.txt                            |   4 ++-
 8 files changed, 50 insertions(+), 4 deletions(-)
 create mode 100644 data/.DS_Store
 create mode 100644 plugins/transform/transform_product_data.py

diff --git a/.DS_Store b/.DS_Store
index fa5394c33c674ace473a5131ed69bbf332f8fb6d..78f7103e15ac2f5b1f853e19876254e73c29aa41 100644
GIT binary patch
literal 8196
zcmeI1&yNx@6vtlyfpKF({E^_nWD>841FLR`7i8Ik2QQ1!gBlzb#K{g*ghAX8k~yn?
zga3k8{X_JB@uc7W5N8I+dfK?LFKJ(=eebpXq@B`%h*)dZuMw4rsE8zVX$RQ`5$z(a
zNk-R{pc(v$JQ_aq`BBg3y)|o7VF(xkhJYbp2p9tY0s)-aLW*0?ebs7mL%<ODFA0eE
zgN-C~Q|m}+`RYI>uK>_FR7*iwTnB`Yt#woDNNKb{m~w?yu1sGsm~uzGt?imxM@lPq
zV*2vIbk9uRP?+o;@oi~Ntf{og4FN-7oq*Ubc{(Ca0liAy-$mO4r0P#8(m6+ca%o7@
z*gap^ecoo*n@`93d%&08*!QNP<9uhCZ0<t7P_PO+g%_RIeBKG&a5inZlV{@TG3S2o
zciHVd@dop5>FPcYLe~oh6Y1df2QYd5)C>B2-r}>MKb5{6)nHjgt6M5B7PZ|*#ol!q
z%Zj~d)SZfbr%_)nt>U$tx9=ZzkNr`=-^-a1)2|k~q@Gv#2b5Oxe%qM@J`aYd6ZWb;
zBsa-}-9XzqH`v8^d_8j>*TKWttJt-@n&44$8nudjuYNZLL`)d6705?4t|p!WlE(t!
zSWw3_ph*gfh#QA;C2mW%<T4!582(duSLp;xZA5-5%8yi(%cn*;J;p_9{~5L12tEyx
z{bs(U`puCKpS7)!5lbUJ%L3-$VrvDM(U?w>QRN2v6x(10*!VO<T;Z{4o5igyI-~={
zXwxI;DthWC<7f9%qbbn@Y~P><PnJe_3Md5eD)M&}CjI3?j`bKJ9M-+uD#*<F4<1Gg
zf%8Zps|H;b_y2?K-~Z3!)XbO-0Yl&~39wwd({907p6VN%#3^qhy+jfs>>{P53$nxy
g0X(n%FhtpgsqkrP9Vv|(i1`p8F__E{I8y?@0Jl~xMF0Q*

delta 167
zcmZp1XfcprU|?W$DortDU=RQ@Ie-{MGjUEV6q~50D9Qqo2Z?#+<R>TP<R^g?Y%E;P
z$jCC;L(qP5xj?Cik)@fAg0Y2Rt&T#qrICS-f{C%&Btgr`UP3~X^Mv&_mhEC$%+A3f
q$PCmA(#{PeTtRkhEd0(qnP0{c<Oq;0Aj=q-Ag%z5ZI0)e!wdlEt{kfX

diff --git a/Dockerfile b/Dockerfile
index 7186eafa..57df9b7b 100644
--- a/Dockerfile
+++ b/Dockerfile
@@ -21,5 +21,6 @@ RUN --mount=type=bind,target=./requirements.txt,src=./requirements.txt \
 COPY --chown=airflow:airflow ../dags /app/dags
 COPY --chown=airflow:airflow ../plugins /app/plugins
 COPY --chown=airflow:airflow ../config /app/config
+COPY --chown=airflow:airflow ../data /app/data
 
 WORKDIR /app
diff --git a/dags/init-db.py b/dags/init-db.py
index b6121c88..fa8494ae 100644
--- a/dags/init-db.py
+++ b/dags/init-db.py
@@ -4,6 +4,7 @@
 from airflow.operators.python import PythonOperator
 
 from plugins.models.initialize import initialize_db
+from plugins.transform.transform_product_data import transform_product_data
 
 
 @dag(
@@ -25,7 +26,12 @@ def initialize():
         python_callable=initialize_db,
     )
 
-    initialize_task
+    transform_product_data_task = PythonOperator(
+        task_id="transform_product_data",
+        python_callable=transform_product_data,
+    )
+
+    initialize_task >> transform_product_data_task
 
 
 initialize()
diff --git a/data/.DS_Store b/data/.DS_Store
new file mode 100644
index 0000000000000000000000000000000000000000..e6dcd11ea903d21b221a1527f175cf68114b349b
GIT binary patch
literal 6148
zcmeHKyH3ME5S)V)K{Tll<^2Gb_yZ?$3JMzd0m!3sFbW9;)%_8l2s5__!ohWf(5|#Q
z>%HB3jO2R&2($g_8khhWQ-wj!fG9lJbW)OqA@|YY30HW*BiezE{$faTzu<f~)HnQ`
z|C%e=VvT1;x7eY=k}G??X}jg3?G~&PY46nPBMUv?40o7giUzmDiv6w*Dy9b&xy+3-
z;0!ne&VVy;GzNHPOUCDhZk+*Vz!~^vK>I_a3hRQ&FmD|+x<0FYB$UwCwOAMn)&-Lx
zTPVRwiB=lQ6C+qT<FVw`1(TtbBjl0UPZ{~^@e=arjK`dgP#d~+2AqLD1E=;m(fL2;
zFEjYa@0a+OGvEyTGX}z_xo>7%R5@FJEl+1{q&`q3%1d%6^m~^8erO-L%#GF_<uk7?
Wm<-j6;<=pY7lBNO8)x7M4159!-Yc;H

literal 0
HcmV?d00001

diff --git a/docker-compose.yaml b/docker-compose.yaml
index 6c48a8c5..f57a3f4f 100644
--- a/docker-compose.yaml
+++ b/docker-compose.yaml
@@ -65,6 +65,8 @@ services:
     environment:
       PGADMIN_DEFAULT_EMAIL: admin@admin.com
       PGADMIN_DEFAULT_PASSWORD: admin
+    networks:
+      - data-network
 
   redis:
     image: redis:latest
@@ -216,6 +218,7 @@ services:
       - ./logs:/sources/logs
       - ./plugins:/sources/plugins
       - ./config:/sources/config
+      - ./data:/sources/data
 
 volumes:
   pgdata:
diff --git a/plugins/models/models.py b/plugins/models/models.py
index 83072c1c..69855b1c 100644
--- a/plugins/models/models.py
+++ b/plugins/models/models.py
@@ -82,7 +82,7 @@ class Order(base):
 
     # relationships
     customerId = Column(
-        Integer, ForeignKey("customer.customerId"), nullable=False
+        String, ForeignKey("customer.customerId"), nullable=False
     )
     customer = relationship("Customer", back_populates="order")
 
@@ -119,8 +119,9 @@ class ShippingCourierStatus(enum.Enum):
 
 class Shipping(base):
     __tablename__ = "shipping"
-    __table_args__ = (PrimaryKeyConstraint("orderId", name="orderId"),)
+    __table_args__ = (PrimaryKeyConstraint("shippingId", name="shippingId"),)
 
+    shippingId = Column(Integer, primary_key=True, autoincrement=True)
     orderId = Column(String, ForeignKey("order.orderId"), nullable=False)
     shipServiceLevel = Column(Enum(ShippingServiceLevel), nullable=False)
     courierStatus = Column(Enum(ShippingCourierStatus), nullable=False)
diff --git a/plugins/transform/transform_product_data.py b/plugins/transform/transform_product_data.py
new file mode 100644
index 00000000..c6f121cf
--- /dev/null
+++ b/plugins/transform/transform_product_data.py
@@ -0,0 +1,33 @@
+import pandas as pd
+from sqlalchemy import create_engine
+
+from config.settings import POSTGRES_CONN_STRING
+
+engine = create_engine(POSTGRES_CONN_STRING)
+
+
+def process_data(csv_file_path, table_name, selected_columns, chunk_size=1000):
+    try:
+        chunks = pd.read_csv(
+            csv_file_path, chunksize=chunk_size, usecols=selected_columns
+        )
+        for chunk in chunks:
+            chunk.to_sql(
+                name=table_name,
+                con=engine,
+                if_exists="append",
+                index=False,
+            )
+            print(
+                f"Data from {csv_file_path} successfully loaded into {table_name}."
+            )
+    except Exception as e:
+        print(f"Error occurred: {e}")
+
+
+def transform_product_data():
+    process_data(
+        "data/amazon-sale-report.csv",
+        "products",
+        ["SKU", "Style", "Category", "Size", "ASIN"],
+    )
diff --git a/requirements.txt b/requirements.txt
index 680c3a39..d2007ffd 100644
--- a/requirements.txt
+++ b/requirements.txt
@@ -86,6 +86,7 @@ methodtools==0.4.7
 more-itertools==10.5.0
 multidict==6.1.0
 nodeenv==1.9.1
+numpy==2.0.2
 opentelemetry-api==1.27.0
 opentelemetry-exporter-otlp==1.27.0
 opentelemetry-exporter-otlp-proto-common==1.27.0
@@ -96,6 +97,7 @@ opentelemetry-sdk==1.27.0
 opentelemetry-semantic-conventions==0.48b0
 ordered-set==4.1.0
 packaging==24.1
+pandas==2.2.3
 pathspec==0.12.1
 pendulum==3.0.0
 platformdirs==4.3.6
@@ -105,7 +107,7 @@ prison==0.2.1
 propcache==0.2.0
 protobuf==4.25.5
 psutil==6.1.0
-psycopg2==2.9.10
+psycopg2-binary==2.9.10
 pycparser==2.22
 Pygments==2.18.0
 PyJWT==2.9.0