# pg_upsert
[](https://github.com/geocoug/pg_upsert/actions/workflows/ci-cd.yml)
[](https://pypi.org/project/pg_upsert/)
[](https://pypi.org/project/pg_upsert/)
**pg_upsert** is a Python package that provides a method to *interactively* update and insert (upsert) rows of a base table or base tables from the staging table(s) of the same name. The package is designed to work exclusively with PostgreSQL databases.
The program will perform initial table checks in the form of not-null, primary key, and foreign key checks. If any of these checks fail, the program will exit with an error message. If all checks pass, the program will display the number of rows to be inserted and updated, and ask for confirmation before proceeding. If the user confirms, the program will perform the upserts and display the number of rows inserted and updated. If the user does not confirm, the program will exit without performing any upserts.
## Credits
This project was created using inspiration from [ExecSQL](https://execsql.readthedocs.io/en/latest/index.html) and the example script [`pg_upsert.sql`](https://osdn.net/projects/execsql-upsert/). The goal of this project is to provide a Python implementation of `pg_upsert.sql` without the need for ExecSQL.
## Installation
1. Create a virtual environment
```sh
python -m venv .venv
```
2. Activate the virtual environment
```sh
source .venv/bin/activate
```
3. Install the package
```sh
pip install pg_upsert
```
## Usage
### CLI
```sh
usage: pg_upsert.py [-h] [-q] [-d] [-l LOGFILE] [-e EXCLUDE_COLUMNS] [-n NULL_COLUMNS] [-c] [-i] [-m METHOD] HOST DATABASE USER STAGING_SCHEMA BASE_SCHEMA TABLE [TABLE ...]
Update and insert (upsert) data from staging tables to base tables.
positional arguments:
HOST database host
DATABASE database name
USER database user
STAGING_SCHEMA staging schema name
BASE_SCHEMA base schema name
TABLE table name(s)
options:
-h, --help show this help message and exit
-q, --quiet suppress all console output
-d, --debug display debug output
-l LOGFILE, --log LOGFILE
write log to LOGFILE
-e EXCLUDE_COLUMNS, --exclude EXCLUDE_COLUMNS
comma-separated list of columns to exclude from null checks
-n NULL_COLUMNS, --null NULL_COLUMNS
comma-separated list of columns to exclude from null checks
-c, --commit commit changes to database
-i, --interactive display interactive GUI of important table information
-m METHOD, --method METHOD
method to use for upsert
```
### Python
```py
import logging
from pathlib import Path
from pg_upsert import upsert
logfile = Path("pg_upsert.log")
if logfile.exists():
logfile.unlink()
logging.basicConfig(
level=logging.INFO,
format="%(message)s",
handlers=[
logging.FileHandler(logfile),
logging.StreamHandler(),
],
)
upsert(
host="localhost",
database="",
user="postgres",
# passwd=, # if not provided, will prompt for password
tables=[],
stg_schema="staging",
base_schema="public",
upsert_method="upsert", # "upsert" | "update" | "insert", default: "upsert"
commit=False, # optional, default=False
interactive=True, # optional, default=False
exclude_cols=[], # optional
exclude_null_check_columns=[], # optional
)
```
### Docker
```sh
docker run --rm -v $(pwd):/app ghcr.io/geocoug/pg_upsert [-h] [-q] [-d] [-l LOGFILE] [-e EXCLUDE_COLUMNS] [-n NULL_COLUMNS] [-c] [-i] [-m METHOD] HOST DATABASE USER STAGING_SCHEMA BASE_SCHEMA TABLE [TABLE ...]
```
## Example
This example will demonstrate how to use `pg_upsert` to upsert data from staging tables to base tables.
1. Initialize a PostgreSQL database called `dev` with the following schema and data.
```sql
-- Create base tables.
drop table if exists public.genres cascade;
create table public.genres (
genre varchar(100) primary key,
description varchar not null
);
drop table if exists public.books cascade;
create table public.books (
book_id varchar(100) primary key,
book_title varchar(200) not null,
genre varchar(100) not null,
notes text,
foreign key (genre) references genres(genre)
);
drop table if exists public.authors cascade;
create table public.authors (
author_id varchar(100) primary key,
first_name varchar(100) not null,
last_name varchar(100) not null
);
drop table if exists public.book_authors cascade;
create table public.book_authors (
book_id varchar(100) not null,
author_id varchar(100) not null,
foreign key (author_id) references authors(author_id),
foreign key (book_id) references books(book_id),
constraint pk_book_authors primary key (book_id, author_id)
);
-- Create staging tables that mimic base tables.
-- Note: staging tables have the same columns as base tables but no PK, FK, or NOT NULL constraints.
create schema if not exists staging;
drop table if exists staging.genres cascade;
create table staging.genres (
genre varchar(100),
description varchar
);
drop table if exists staging.books cascade;
create table staging.books (
book_id varchar(100),
book_title varchar(200),
genre varchar(100),
notes text
);
drop table if exists staging.authors cascade;
create table staging.authors (
author_id varchar(100),
first_name varchar(100),
last_name varchar(100)
);
drop table if exists staging.book_authors cascade;
create table staging.book_authors (
book_id varchar(100),
author_id varchar(100)
);
-- Insert data into staging tables.
insert into staging.genres (genre, description) values
('Fiction', 'Literary works that are imaginary, not based on real events or people'),
('Non-Fiction', 'Literary works based on real events, people, and facts');
insert into staging.authors (author_id, first_name, last_name) values
('JDoe', 'John', 'Doe'),
('JSmith', 'Jane', 'Smith'),
('JTrent', 'Joe', 'Trent');
insert into staging.books (book_id, book_title, genre, notes) values
('B001', 'The Great Novel', 'Fiction', 'An epic tale of love and loss'),
('B002', 'Not Another Great Novel', 'Non-Fiction', 'A comprehensive guide to writing a great novel');
insert into staging.book_authors (book_id, author_id) values
('B001', 'JDoe'),
('B001', 'JTrent'),
('B002', 'JSmith');
```
2. Create a Python script called `upsert_data.py` that calls `pg_upsert` to upsert data from staging tables to base tables.
```py
import logging
from pathlib import Path
from pg_upsert import upsert
logfile = Path("pg_upsert.log")
if logfile.exists():
logfile.unlink()
logging.basicConfig(
level=logging.INFO,
format="%(message)s",
handlers=[
logging.FileHandler(logfile),
logging.StreamHandler(),
],
)
upsert(
host="localhost",
database="dev",
user="docker", # Change this
tables=["books", "authors", "genres", "book_authors"],
stg_schema="staging",
base_schema="public",
upsert_method="upsert",
commit=True,
interactive=False,
exclude_cols=[],
exclude_null_check_columns=[],
)
```
3. Run the script: `python upsert_data.py`
```txt
The script pg_upsert.py wants the password for PostgresDB(host=localhost, database=dev, user=docker):
Upserting to public from staging
Tables selected for upsert:
books
authors
genres
book_authors
===Non-NULL checks===
Conducting non-null QA checks on table staging.books
Conducting non-null QA checks on table staging.authors
Conducting non-null QA checks on table staging.genres
Conducting non-null QA checks on table staging.book_authors
===Primary Key checks===
Conducting primary key QA checks on table staging.books
Conducting primary key QA checks on table staging.authors
Conducting primary key QA checks on table staging.genres
Conducting primary key QA checks on table staging.book_authors
===Foreign Key checks===
Conducting foreign key QA checks on table staging.books
Conducting foreign key QA checks on table staging.authors
Conducting foreign key QA checks on table staging.genres
Conducting foreign key QA checks on table staging.book_authors
===QA checks passed. Starting upsert===
Performing upsert on table public.genres
Adding data to public.genres
2 rows inserted
Performing upsert on table public.authors
Adding data to public.authors
3 rows inserted
Performing upsert on table public.books
Adding data to public.books
2 rows inserted
Performing upsert on table public.book_authors
Adding data to public.book_authors
3 rows inserted
Changes committed
```
4. Modify a row in the staging table.
```sql
update staging.books set book_title = 'The Great Novel 2' where book_id = 'B001';
```
5. Run the script again, but this time set `interactive=True` in the `upsert` function call in `upsert_data.py`.
The script will display GUI dialogs during the upsert process to show which rows will be added and which rows will be updated. The user can chose to confirm, skip, or cancel the upsert process at any time. The script will not commit any changes to the database until all of the upserts have been completed successfully.
6. Let's modify the `staging.books` table to include a row with a missing value in the `book_title` and `Mystery` value in the `genre` column to see what happens.
```sql
insert into staging.books (book_id, book_title, genre, notes)
values ('B003', null, 'Mystery', 'A book with no name!');
```
Run the script again: `python upsert_data.py`
```txt
The script pg_upsert.py wants the password for PostgresDB(host=localhost, database=dev, user=docker):
Upserting to public from staging
===Non-NULL checks===
Conducting non-null QA checks on table staging.books
Column book_title has 1 null values
Conducting non-null QA checks on table staging.authors
Conducting non-null QA checks on table staging.genres
Conducting non-null QA checks on table staging.book_authors
===Primary Key checks===
Conducting primary key QA checks on table staging.books
Conducting primary key QA checks on table staging.authors
Conducting primary key QA checks on table staging.genres
Conducting primary key QA checks on table staging.book_authors
===Foreign Key checks===
Conducting foreign key QA checks on table staging.books
Foreign key error referencing genres
Conducting foreign key QA checks on table staging.authors
Conducting foreign key QA checks on table staging.genres
Conducting foreign key QA checks on table staging.book_authors
QA checks failed. Aborting upsert.
```
The script failed to upsert data because there are non-null and foreign key checks that failed on the `staging.books` table. The interactive GUI will display all values in the `books.genres` column that fail the foreign key check. No GUI dialogs are displayed for non-null checks, because there are no values to display. Similarly, if there is a primary key check that fails, a GUI dialog will be displayed with the primary keys in the table that are failing.
Raw data
{
"_id": null,
"home_page": null,
"name": "pg-upsert",
"maintainer": null,
"docs_url": null,
"requires_python": ">=3.8",
"maintainer_email": null,
"keywords": "postgresql, postgres, dbms, etl, upsert, database",
"author": null,
"author_email": "Caleb Grant <grantcaleb22@gmail.com>",
"download_url": "https://files.pythonhosted.org/packages/f5/3d/c1228973b1a56f2d6750591ddf298c8b8def52eb6c62e30b4446ff3f1a38/pg_upsert-0.0.9.tar.gz",
"platform": null,
"description": "# pg_upsert\n\n[](https://github.com/geocoug/pg_upsert/actions/workflows/ci-cd.yml)\n[](https://pypi.org/project/pg_upsert/)\n[](https://pypi.org/project/pg_upsert/)\n\n**pg_upsert** is a Python package that provides a method to *interactively* update and insert (upsert) rows of a base table or base tables from the staging table(s) of the same name. The package is designed to work exclusively with PostgreSQL databases.\n\nThe program will perform initial table checks in the form of not-null, primary key, and foreign key checks. If any of these checks fail, the program will exit with an error message. If all checks pass, the program will display the number of rows to be inserted and updated, and ask for confirmation before proceeding. If the user confirms, the program will perform the upserts and display the number of rows inserted and updated. If the user does not confirm, the program will exit without performing any upserts.\n\n## Credits\n\nThis project was created using inspiration from [ExecSQL](https://execsql.readthedocs.io/en/latest/index.html) and the example script [`pg_upsert.sql`](https://osdn.net/projects/execsql-upsert/). The goal of this project is to provide a Python implementation of `pg_upsert.sql` without the need for ExecSQL.\n\n## Installation\n\n1. Create a virtual environment\n\n ```sh\n python -m venv .venv\n ```\n\n2. Activate the virtual environment\n\n ```sh\n source .venv/bin/activate\n ```\n\n3. Install the package\n\n ```sh\n pip install pg_upsert\n ```\n\n## Usage\n\n### CLI\n\n```sh\nusage: pg_upsert.py [-h] [-q] [-d] [-l LOGFILE] [-e EXCLUDE_COLUMNS] [-n NULL_COLUMNS] [-c] [-i] [-m METHOD] HOST DATABASE USER STAGING_SCHEMA BASE_SCHEMA TABLE [TABLE ...]\n\nUpdate and insert (upsert) data from staging tables to base tables.\n\npositional arguments:\n HOST database host\n DATABASE database name\n USER database user\n STAGING_SCHEMA staging schema name\n BASE_SCHEMA base schema name\n TABLE table name(s)\n\noptions:\n -h, --help show this help message and exit\n -q, --quiet suppress all console output\n -d, --debug display debug output\n -l LOGFILE, --log LOGFILE\n write log to LOGFILE\n -e EXCLUDE_COLUMNS, --exclude EXCLUDE_COLUMNS\n comma-separated list of columns to exclude from null checks\n -n NULL_COLUMNS, --null NULL_COLUMNS\n comma-separated list of columns to exclude from null checks\n -c, --commit commit changes to database\n -i, --interactive display interactive GUI of important table information\n -m METHOD, --method METHOD\n method to use for upsert\n```\n\n### Python\n\n```py\nimport logging\nfrom pathlib import Path\n\nfrom pg_upsert import upsert\n\n\nlogfile = Path(\"pg_upsert.log\")\nif logfile.exists():\n logfile.unlink()\n\nlogging.basicConfig(\n level=logging.INFO,\n format=\"%(message)s\",\n handlers=[\n logging.FileHandler(logfile),\n logging.StreamHandler(),\n ],\n)\n\nupsert(\n host=\"localhost\",\n database=\"\",\n user=\"postgres\",\n # passwd=, # if not provided, will prompt for password\n tables=[],\n stg_schema=\"staging\",\n base_schema=\"public\",\n upsert_method=\"upsert\", # \"upsert\" | \"update\" | \"insert\", default: \"upsert\"\n commit=False, # optional, default=False\n interactive=True, # optional, default=False\n exclude_cols=[], # optional\n exclude_null_check_columns=[], # optional\n)\n```\n\n### Docker\n\n```sh\ndocker run --rm -v $(pwd):/app ghcr.io/geocoug/pg_upsert [-h] [-q] [-d] [-l LOGFILE] [-e EXCLUDE_COLUMNS] [-n NULL_COLUMNS] [-c] [-i] [-m METHOD] HOST DATABASE USER STAGING_SCHEMA BASE_SCHEMA TABLE [TABLE ...]\n```\n\n## Example\n\nThis example will demonstrate how to use `pg_upsert` to upsert data from staging tables to base tables.\n\n1. Initialize a PostgreSQL database called `dev` with the following schema and data.\n\n ```sql\n -- Create base tables.\n drop table if exists public.genres cascade;\n create table public.genres (\n genre varchar(100) primary key,\n description varchar not null\n );\n\n drop table if exists public.books cascade;\n create table public.books (\n book_id varchar(100) primary key,\n book_title varchar(200) not null,\n genre varchar(100) not null,\n notes text,\n foreign key (genre) references genres(genre)\n );\n\n drop table if exists public.authors cascade;\n create table public.authors (\n author_id varchar(100) primary key,\n first_name varchar(100) not null,\n last_name varchar(100) not null\n );\n\n drop table if exists public.book_authors cascade;\n create table public.book_authors (\n book_id varchar(100) not null,\n author_id varchar(100) not null,\n foreign key (author_id) references authors(author_id),\n foreign key (book_id) references books(book_id),\n constraint pk_book_authors primary key (book_id, author_id)\n );\n\n -- Create staging tables that mimic base tables.\n -- Note: staging tables have the same columns as base tables but no PK, FK, or NOT NULL constraints.\n create schema if not exists staging;\n\n drop table if exists staging.genres cascade;\n create table staging.genres (\n genre varchar(100),\n description varchar\n );\n\n drop table if exists staging.books cascade;\n create table staging.books (\n book_id varchar(100),\n book_title varchar(200),\n genre varchar(100),\n notes text\n );\n\n drop table if exists staging.authors cascade;\n create table staging.authors (\n author_id varchar(100),\n first_name varchar(100),\n last_name varchar(100)\n );\n\n drop table if exists staging.book_authors cascade;\n create table staging.book_authors (\n book_id varchar(100),\n author_id varchar(100)\n );\n\n -- Insert data into staging tables.\n insert into staging.genres (genre, description) values\n ('Fiction', 'Literary works that are imaginary, not based on real events or people'),\n ('Non-Fiction', 'Literary works based on real events, people, and facts');\n\n insert into staging.authors (author_id, first_name, last_name) values\n ('JDoe', 'John', 'Doe'),\n ('JSmith', 'Jane', 'Smith'),\n ('JTrent', 'Joe', 'Trent');\n\n insert into staging.books (book_id, book_title, genre, notes) values\n ('B001', 'The Great Novel', 'Fiction', 'An epic tale of love and loss'),\n ('B002', 'Not Another Great Novel', 'Non-Fiction', 'A comprehensive guide to writing a great novel');\n\n insert into staging.book_authors (book_id, author_id) values\n ('B001', 'JDoe'),\n ('B001', 'JTrent'),\n ('B002', 'JSmith');\n ```\n\n2. Create a Python script called `upsert_data.py` that calls `pg_upsert` to upsert data from staging tables to base tables.\n\n ```py\n import logging\n from pathlib import Path\n\n from pg_upsert import upsert\n\n logfile = Path(\"pg_upsert.log\")\n if logfile.exists():\n logfile.unlink()\n\n logging.basicConfig(\n level=logging.INFO,\n format=\"%(message)s\",\n handlers=[\n logging.FileHandler(logfile),\n logging.StreamHandler(),\n ],\n )\n\n upsert(\n host=\"localhost\",\n database=\"dev\",\n user=\"docker\", # Change this\n tables=[\"books\", \"authors\", \"genres\", \"book_authors\"],\n stg_schema=\"staging\",\n base_schema=\"public\",\n upsert_method=\"upsert\",\n commit=True,\n interactive=False,\n exclude_cols=[],\n exclude_null_check_columns=[],\n )\n ```\n\n3. Run the script: `python upsert_data.py`\n\n ```txt\n The script pg_upsert.py wants the password for PostgresDB(host=localhost, database=dev, user=docker):\n Upserting to public from staging\n Tables selected for upsert:\n books\n authors\n genres\n book_authors\n\n ===Non-NULL checks===\n Conducting non-null QA checks on table staging.books\n Conducting non-null QA checks on table staging.authors\n Conducting non-null QA checks on table staging.genres\n Conducting non-null QA checks on table staging.book_authors\n\n ===Primary Key checks===\n Conducting primary key QA checks on table staging.books\n Conducting primary key QA checks on table staging.authors\n Conducting primary key QA checks on table staging.genres\n Conducting primary key QA checks on table staging.book_authors\n\n ===Foreign Key checks===\n Conducting foreign key QA checks on table staging.books\n Conducting foreign key QA checks on table staging.authors\n Conducting foreign key QA checks on table staging.genres\n Conducting foreign key QA checks on table staging.book_authors\n\n ===QA checks passed. Starting upsert===\n Performing upsert on table public.genres\n Adding data to public.genres\n 2 rows inserted\n Performing upsert on table public.authors\n Adding data to public.authors\n 3 rows inserted\n Performing upsert on table public.books\n Adding data to public.books\n 2 rows inserted\n Performing upsert on table public.book_authors\n Adding data to public.book_authors\n 3 rows inserted\n\n Changes committed\n ```\n\n4. Modify a row in the staging table.\n\n ```sql\n update staging.books set book_title = 'The Great Novel 2' where book_id = 'B001';\n ```\n\n5. Run the script again, but this time set `interactive=True` in the `upsert` function call in `upsert_data.py`.\n\n The script will display GUI dialogs during the upsert process to show which rows will be added and which rows will be updated. The user can chose to confirm, skip, or cancel the upsert process at any time. The script will not commit any changes to the database until all of the upserts have been completed successfully.\n\n \n\n6. Let's modify the `staging.books` table to include a row with a missing value in the `book_title` and `Mystery` value in the `genre` column to see what happens.\n\n ```sql\n insert into staging.books (book_id, book_title, genre, notes)\n values ('B003', null, 'Mystery', 'A book with no name!');\n ```\n\n Run the script again: `python upsert_data.py`\n\n ```txt\n The script pg_upsert.py wants the password for PostgresDB(host=localhost, database=dev, user=docker):\n Upserting to public from staging\n\n ===Non-NULL checks===\n Conducting non-null QA checks on table staging.books\n Column book_title has 1 null values\n Conducting non-null QA checks on table staging.authors\n Conducting non-null QA checks on table staging.genres\n Conducting non-null QA checks on table staging.book_authors\n\n ===Primary Key checks===\n Conducting primary key QA checks on table staging.books\n Conducting primary key QA checks on table staging.authors\n Conducting primary key QA checks on table staging.genres\n Conducting primary key QA checks on table staging.book_authors\n\n ===Foreign Key checks===\n Conducting foreign key QA checks on table staging.books\n Foreign key error referencing genres\n Conducting foreign key QA checks on table staging.authors\n Conducting foreign key QA checks on table staging.genres\n Conducting foreign key QA checks on table staging.book_authors\n\n QA checks failed. Aborting upsert.\n ```\n\n The script failed to upsert data because there are non-null and foreign key checks that failed on the `staging.books` table. The interactive GUI will display all values in the `books.genres` column that fail the foreign key check. No GUI dialogs are displayed for non-null checks, because there are no values to display. Similarly, if there is a primary key check that fails, a GUI dialog will be displayed with the primary keys in the table that are failing.\n",
"bugtrack_url": null,
"license": null,
"summary": "A Python library for upserting data into postgres.",
"version": "0.0.9",
"project_urls": {
"Homepage": "https://github.com/geocoug/pg_upsert",
"Issues": "https://github.com/geocoug/pg_upsert/issues"
},
"split_keywords": [
"postgresql",
" postgres",
" dbms",
" etl",
" upsert",
" database"
],
"urls": [
{
"comment_text": "",
"digests": {
"blake2b_256": "9d19e98904cac8ef3d57be960f145b6df5d60ae2df4116f36ac792165d6f3f81",
"md5": "0201835c943f67eee6fbfc4ddd6149ab",
"sha256": "376d781450a160fba8c6df3a36a40537441ce7767006cb4288d9f93bef1487b7"
},
"downloads": -1,
"filename": "pg_upsert-0.0.9-py3-none-any.whl",
"has_sig": false,
"md5_digest": "0201835c943f67eee6fbfc4ddd6149ab",
"packagetype": "bdist_wheel",
"python_version": "py3",
"requires_python": ">=3.8",
"size": 33259,
"upload_time": "2024-06-10T22:05:37",
"upload_time_iso_8601": "2024-06-10T22:05:37.502976Z",
"url": "https://files.pythonhosted.org/packages/9d/19/e98904cac8ef3d57be960f145b6df5d60ae2df4116f36ac792165d6f3f81/pg_upsert-0.0.9-py3-none-any.whl",
"yanked": false,
"yanked_reason": null
},
{
"comment_text": "",
"digests": {
"blake2b_256": "f53dc1228973b1a56f2d6750591ddf298c8b8def52eb6c62e30b4446ff3f1a38",
"md5": "0915f5cb475d5b1c70bfd0d3e6a2a67d",
"sha256": "5474d95c07f99ed3de1c12387ccd30a508521df02a09fd3413d15cbeca023676"
},
"downloads": -1,
"filename": "pg_upsert-0.0.9.tar.gz",
"has_sig": false,
"md5_digest": "0915f5cb475d5b1c70bfd0d3e6a2a67d",
"packagetype": "sdist",
"python_version": "source",
"requires_python": ">=3.8",
"size": 36776,
"upload_time": "2024-06-10T22:05:39",
"upload_time_iso_8601": "2024-06-10T22:05:39.186617Z",
"url": "https://files.pythonhosted.org/packages/f5/3d/c1228973b1a56f2d6750591ddf298c8b8def52eb6c62e30b4446ff3f1a38/pg_upsert-0.0.9.tar.gz",
"yanked": false,
"yanked_reason": null
}
],
"upload_time": "2024-06-10 22:05:39",
"github": true,
"gitlab": false,
"bitbucket": false,
"codeberg": false,
"github_user": "geocoug",
"github_project": "pg_upsert",
"travis_ci": false,
"coveralls": false,
"github_actions": true,
"requirements": [],
"lcname": "pg-upsert"
}
JSON
