Refactor database.py to simplify table creation and property insertion logic

- Replace complex table creation and insertion queries with more focused methods
- Create a new `create_properties_table` function for table initialization
- Modify `get_new_properties` to use LEFT JOIN for filtering new properties
- Add `insert_new_properties` method to handle property insertion
- Simplify database operations and improve code readability

Author: Adriano Sanges

real-estate-etl/database.py

@@ -2,60 +2,39 @@
 import polars as pl
 import duckdb
 
-def clean_properties(con: duckdb.DuckDBPyConnection) -> None:
-    logging.debug("Starting property cleaning")
-    create_table_query = """
-CREATE TABLE IF NOT EXISTS main.cleaned_properties (
-    url TEXT PRIMARY KEY,
-    title TEXT,
-    content TEXT,
-    price INTEGER,
-    city TEXT,
-    neighbourhood TEXT,
-    road TEXT,
-    square_meters INTEGER,
-    floor TEXT,
-    garage_info TEXT
-);
-CREATE OR REPLACE TABLE main.new_properties (
-    url TEXT PRIMARY KEY,
-    title TEXT,
-    content TEXT,
-    price INTEGER,
-    city TEXT,
-    neighbourhood TEXT,
-    road TEXT,
-    square_meters INTEGER,
-    floor TEXT,
-    garage_info TEXT
-);
-"""
-    con.sql(create_table_query)
-    insert_query = """
-INSERT INTO main.cleaned_properties (url, title, price, city, neighbourhood, road, square_meters, floor, garage_info)
-SELECT url, title, price, city, neighbourhood, road, square_meters, floor, garage_info
-FROM main.properties
-WHERE NOT EXISTS (  
-    SELECT 1 
-    FROM main.cleaned_properties
-    WHERE main.cleaned_properties.url = main.properties.url
-);
-"""
-    insert_query_only_new = """
-INSERT INTO main.new_properties (url, title, price, city, neighbourhood, road, square_meters, floor, garage_info)
-SELECT url, title, price, city, neighbourhood, road, square_meters, floor, garage_info
-FROM main.properties
-WHERE NOT EXISTS (  
-    SELECT 1 
-    FROM main.cleaned_properties
-    WHERE main.cleaned_properties.url = main.properties.url
-)
-"""
-    con.sql(insert_query_only_new)
-    con.sql(insert_query)
-    con.sql("DELETE FROM main.properties;")
+
+def create_properties_table(con: duckdb.DuckDBPyConnection) -> None:
+        # Create the table 'properties'
+    con.execute("""
+        CREATE TABLE IF NOT EXISTS main.properties (
+            url VARCHAR PRIMARY KEY,
+            title VARCHAR,
+            content VARCHAR[],    -- storing list[str] as an array of VARCHAR
+            price BIGINT,
+            road VARCHAR,
+            square_meters BIGINT,
+            floor BIGINT,
+            garage_info VARCHAR
+        )
+    """)
 
 
 def get_new_properties(con: duckdb.DuckDBPyConnection) -> pl.DataFrame:
-    df = con.sql("SELECT * FROM main.new_properties;").pl()
-    return df
+        # Use a LEFT JOIN to filter out rows that already exist in 'properties'
+    new_rows_df = con.execute("""
+        SELECT nd.*
+        FROM new_data nd
+        LEFT JOIN main.properties p ON nd.url = p.url
+        WHERE p.url IS NULL
+    """).pl()
+
+
+def insert_new_properties(con: duckdb.DuckDBPyConnection) -> None:
+        # Insert the new rows into the 'properties' table
+    con.execute("""
+        INSERT INTO properties
+        SELECT nd.*
+        FROM new_data nd
+        LEFT JOIN main.properties p ON nd.url = p.url
+        WHERE p.url IS NULL
+    """)

real-estate-etl/pyproject.toml

@@ -7,6 +7,8 @@ requires-python = ">=3.11"
 dependencies = [
     "beautifulsoup4>=4.12.3",
     "duckdb>=1.1.3",
+    "numpy>=2.2.2",
+    "pandas>=2.2.3",
     "polars>=1.21.0",
     "pyarrow>=19.0.0",
     "python-dotenv>=1.0.1",

real-estate-etl/scan_properties.py

@@ -2,7 +2,7 @@
 import polars as pl
 import duckdb
 from scraper import parse_listing
-from database import clean_properties, get_new_properties
+from database import create_properties_table, get_new_properties, insert_new_properties
 from dotenv import load_dotenv
 import os
 from telegram_api import send_message, format_property_message
@@ -19,18 +19,24 @@
     motherduck_token: str = os.getenv("motherduck_token")
 
     data: list[dict] = parse_listing(url)
-
     polars_df: pl.DataFrame = pl.DataFrame(data)
 
     con: duckdb.DuckDBPyConnection = duckdb.connect(f"md:{warehouse_name}?motherduck_token={motherduck_token}")
+
+    create_properties_table(con)
 
-    con.sql("create table if not exists main.properties as select * from polars_df")
+    con.register("new_data", polars_df)
+
+    new_rows_df: pl.DataFrame = get_new_properties(con)
+
 
-    clean_properties(con)
+    insert_new_properties(con)
 
-    new_properties: pl.DataFrame = get_new_properties(con)
     # Iterate over the DataFrame and format each property
-    messages: list[str] = [format_property_message(row) for row in new_properties.iter_rows(named=True)]
+    if new_rows_df is not None:
+        messages: list[str] = [format_property_message(row) for row in new_rows_df.iter_rows(named=True)]
+    else:
+        messages = []
 
 # Send messages in chunks of two
     for i in range(0, len(messages), 2):