Postgres replication

pyproject.toml

@@ -32,6 +32,7 @@ black = "^23.3.0"
 pypdf2 = "^3.0.1"
 greenlet = "<3.0.0"
 confluent-kafka = "^2.3.0"
+types-psycopg2 = "^2.9.0"
 pytest-mock = "^3.12.0"
 twisted = "22.10.0"
 pytest-forked = "^1.6.0"
@@ -40,6 +41,10 @@ pytest-forked = "^1.6.0"
 sqlalchemy = ">=1.4"
 pymysql = "^1.0.3"
 
+[tool.poetry.group.pg_replication.dependencies]
+psycopg2 = ">=2.9.9"
+pypgoutput = "0.0.3"
+
 [tool.poetry.group.google_sheets.dependencies]
 google-api-python-client = "^2.78.0"
 

sources/.dlt/example.secrets.toml

@@ -19,4 +19,4 @@ location = "US"
 ## chess pipeline
 # the section below defines secrets for "chess_dlt_config_example" source in chess/__init__.py
 [sources.chess]
-secret_str="secret string"  # a string secret
+secret_str="secret string"  # a string secret

sources/pg_replication/README.md

@@ -0,0 +1,67 @@
+# Postgres replication
+[Postgres](https://www.postgresql.org/) is one of the most popular relational database management systems. This verified source uses Postgres' replication functionality to efficiently process changes in tables (a process often referred to as _Change Data Capture_ or CDC). It uses [logical decoding](https://www.postgresql.org/docs/current/logicaldecoding.html) and the standard built-in `pgoutput` [output plugin](https://www.postgresql.org/docs/current/logicaldecoding-output-plugin.html).
+
+Resources that can be loaded using this verified source are:
+
+| Name                 | Description                                     |
+|----------------------|-------------------------------------------------|
+| replication_resource | Load published messages from a replication slot |
+
+## Initialize the pipeline
+
+```bash
+dlt init pg_replication duckdb
+```
+
+This uses `duckdb` as destination, but you can choose any of the supported [destinations](https://dlthub.com/docs/dlt-ecosystem/destinations/).
+
+## Add `sql_database` source
+
+```bash
+dlt init sql_database duckdb
+```
+
+This source depends on the [sql_database](../sql_database/README.md) verified source internally to perform initial loads. This step can be skipped if you don't do initial loads.
+## Set up user
+
+The Postgres user needs to have the `LOGIN` and `REPLICATION` attributes assigned:
+
+```sql
+CREATE ROLE replication_user WITH LOGIN REPLICATION;
+```
+
+It also needs `CREATE` privilege on the database:
+
+```sql
+GRANT CREATE ON DATABASE dlt_data TO replication_user;
+```
+
+## Add credentials
+1. Open `.dlt/secrets.toml`.
+2. Enter your Postgres credentials:
+
+    ```toml
+    [sources.pg_replication]
+    credentials="postgresql://replication_user:<<password>>@localhost:5432/dlt_data"
+    ```
+3. Enter credentials for your chosen destination as per the [docs](https://dlthub.com/docs/dlt-ecosystem/destinations/).
+
+## Run the pipeline
+
+1. Install the necessary dependencies by running the following command:
+
+   ```bash
+   pip install -r requirements.txt
+   ```
+
+1. Now the pipeline can be run by using the command:
+
+   ```bash
+   python pg_replication_pipeline.py
+   ```
+
+1. To make sure that everything is loaded as expected, use the command:
+
+   ```bash
+   dlt pipeline pg_replication_pipeline show
+   ```