Warum? Weil ich wissen wollte, ob es geht, aber das kennen wir ja. Für die meisten Anwendungsfälle mag das keine gute Idee sein, aber es macht auf jeden Fall Spaß.
Docker-Image und Installation der Extension
Das wichtigste Element ist die Postgres-Extension pgsql-http, die es uns erlaubt, HTTP-Aufrufe direkt aus Postgres heraus zu machen. Diese Extension ist nicht standardmäßig enthalten, daher müssen wir ein eigenes Docker-Image erstellen, das diese Abhängigkeit beinhaltet. Ein einfaches Dockerfile wie das folgende reicht aus:
FROM postgres:17
RUN apt-get update \
&& apt-get install -y curl \
&& apt-get install -y libcurl4-openssl-dev make gcc g++ \
&& apt-get install -y postgresql-server-dev-17 \
&& apt-get -y install postgresql-17-http
Hinweis: Falls postgresql-17-http nicht direkt verfügbar ist, kann die Extension auch manuell über PGXN oder durch Kompilieren installiert werden. Um die HTTP-Extension zu kompilieren, sind zusätzliche Pakete erforderlich.
Sobald du das Image gestartet hast, musst du die Extension laden. Du kannst die Tabelle pg_extension überprüfen, um sicherzustellen, dass die Extension erfolgreich installiert wurde.
CREATE SCHEMA IF NOT EXISTS scrapedb; -- Stelle sicher, dass das Schema existiert
CREATE EXTENSION http SCHEMA scrapedb;
SELECT * FROM pg_extension;
Grundlegende Tabellen und Typen
Um einen einfachen Scraper zu implementieren, benötigen wir ein paar Dinge:
- Entrypoint: Eine URL, wo das Scraping beginnt.
- Tasks: Eine Liste von Aufgaben, die der Scraper abarbeiten soll. Jede Aufgabe sollte einen Status haben, wie
available,in_flight,finishedoderfailed. - Title: Ein Ort, um die Ergebnisse (Filmtitel) zu speichern, die wir scrapen wollen.
- Liste von Mustern und Handlern: Damit wir wissen, wie wir jede Task-URL mit dem Status
availablebearbeiten.
Erstellung der Tabellen
CREATE TYPE status_enum AS ENUM ('failed', 'available', 'in_flight', 'finished');
CREATE TABLE task (
id SERIAL PRIMARY KEY,
url TEXT NOT NULL,
status status_enum NOT NULL DEFAULT 'available',
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
);
CREATE TABLE title (
id SERIAL PRIMARY KEY,
original_title TEXT,
original_release_year INT,
director TEXT,
object_type TEXT NOT NULL
);
Damit Scrapes effizient durchgeführt werden können, sollten wir Indizes auf häufig verwendete Spalten setzen:
CREATE INDEX idx_task_status ON task(status);
CREATE INDEX idx_task_url ON task(url);
Scraping-Mechanismus
Scrape-Start
Wir wollen eine Funktion, um Scraping-Aufträge zu starten:
CREATE OR REPLACE FUNCTION start_scrape(url_input TEXT)
RETURNS VOID AS $$
BEGIN
INSERT INTO task (url, status)
VALUES (url_input, 'available');
END;
$$
LANGUAGE plpgsql;
HTTP-Anfrage verarbeiten
BEGIN
FOR rec IN
SELECT *
FROM http_get(task_url)
LOOP
IF rec.status = 200 THEN
UPDATE task SET status = 'finished' WHERE url = task_url;
RETURN QUERY SELECT rec.content::jsonb, rec.status;
ELSE
UPDATE task SET status = 'failed' WHERE url = task_url;
RETURN QUERY SELECT '{}'::jsonb, rec.status;
END IF;
END LOOP;
EXCEPTION WHEN OTHERS THEN
UPDATE task SET status = 'failed' WHERE url = task_url;
RETURN QUERY SELECT '{}'::jsonb, 500; -- Hier könnte eine spezifische Fehlerbehandlung eingeführt werden (z.B. für 4xx/5xx Fehlercodes).
END;
END;
$$
LANGUAGE plpgsql;
Automatische Verarbeitung neuer Tasks
Prozess zur Bearbeitung neuer Tasks:
CREATE OR REPLACE FUNCTION process_new_task()
RETURNS TRIGGER AS $$
DECLARE
rec RECORD;
payload JSONB;
BEGIN
IF NEW.status = 'available' AND NEW.url LIKE '%/apis.example.com/titles/top_x/object_type%' THEN
WITH resp AS (
SELECT content AS resp_body, status FROM process_task(NEW.url)
), extract_ids AS (
SELECT array_agg((elem->>'id')::int) AS ids
FROM (SELECT jsonb_array_elements(resp_body) AS elem FROM resp) subquery
), tasks AS (
SELECT 'https://apis.example.com/object_type/movie?id='|| unnest(ids) AS url
FROM extract_ids
)
INSERT INTO task (url, status)
SELECT url, 'available'::status_enum FROM tasks;
ELSIF NEW.status = 'available' AND NEW.url LIKE '%/apis.example.com/object_type%' THEN
FOR rec IN
SELECT content::jsonb AS payload, status FROM process_task(NEW.url)
LOOP
payload := rec.payload;
INSERT INTO title (
original_title, original_release_year, director, object_type
)
SELECT
payload->>'original_title',
(payload->>'original_release_year')::INT,
payload->>'director',
payload->>'object_type'
ON CONFLICT (id) DO NOTHING; -- Hier könnte eine Strategie zur Fehlerbehandlung bei Konflikten eingeführt werden.
END LOOP;
END IF;
RETURN NEW; -- Rückgabe des neuen Datensatzes.
END;
$$
LANGUAGE plpgsql;