How to: Postgres

[K0nkere]

Получить перечень таблиц в базе данных

query = """
SELECT *
FROM pg_catalog.pg_tables
WHERE schemaname != 'pg_catalog' AND
    schemaname != 'information_schema';
"""
pd.read_sql(query, con=engine)

[K0nkere]

Считать таблицу из PostgreSQL в pandas DataFrame

import pandas as pd
import os
import sqlalchemy

# %env PG_USER=root # setting env vars out of Jupyter Notebook
# %env PG_PASS=root

user = os.getenv("PG_USER")
pw = os.getenv("PG_PASS")

engine = sqlalchemy.create_engine(f"postgresql://{user}:{pw}@localhost/ny_taxi")

engine.connect()

query = """
SELECT VendorID, pulocationid, SUM(total_amount)
    FROM green_taxi_2020
GROUP BY VendorID, pulocationid
"""

pd.read_sql(query, con=engine, index_col="index")

Подгрузка батчами

batch_size=<>
batch_gen = pd.read_sql(query, con=engine, chunksize=batch_size)
next(batch_gen)

[K0nkere]

Импорт данных из кластера Yandex Managed Service for PostgreSQL с помощью Sqoop

[K0nkere]

Postgres roles management

Log in pgAdmin >> pg-database >> Create a new role dbt_production with a password and can login create roles create databases privileges, all other settings are default test login via pgcli

pgcli -h localhost -u dbt_production -p 5432 -d ny_taxi

Granting privileges

pgAdmin: go to pg-database >>ny_taxi >> pg-database >> Schemas create or select a new schema de_dbt_production >> Properties >> Securities and Grant privileges for dbt_production user It is possible to do the same via pgCLI or pgAdmin query under the root login

CREATE USER dbt_production WITH PASSWORD 'userpass';
GRANT ALL ON SCHEMA de_dbt_production TO dbt_production;

Queries from user in transaction mode

schema == de_dbt_production

query = """
CREATE TABLE IF NOT EXISTS <schema>.test_db (
    id uuid default fn_uuid_time_ordered() primary key,
    email varchar(300) unique,
    name text)
;

### !!! Granting roles for the table on creating !!!
GRANT ALL ON TABLE <schema>.test_db TO dbt_production;

INSERT INTO
    <schema>.test_db (
        email,
        name
    )
VALUES
    ('email@email51.com', 'Mihael'),
    ('email@email52.com', 'Seva'),
;
"""

connect.execution_options(isolation_level="READ COMMITTED")
cursor = connect.begin()

try: 
    connect.execute(query)
    cursor.commit()
except:
    print("Rollback of transaction")
    cursor.rollback()

cursor.close()

Reading data

query = """
SELECT * FROM <schema>.test_db
"""
connect.execution_options(isolation_level="READ COMMITTED")
cursor = connect.begin()

try:
    result = connect.execute(query)
    cursor.commit()

except:
    print("Rollback of transaction")
    cursor.rollback()

cursor.close()
result.fetchall()

[K0nkere]

Обновление записи в БД или добавление в случае отсутствия Вариант 1 - подходит для версионирования на основе valid_from/_to

update public.vip_users_channels
    set valid_to = current_timestamp
    where valid_from <= current_timestamp 
        and valid_to is null 
        and email='{user_email}'
        and channel_type = '{channel_type}'
;
insert into public.vip_users_channels(email, channel_type, author_id, conversation_id, conversation_name)
values ('{user_email}', '{channel_type}', '{author_id}', '{conversation_id}', '{conversation_name}');

Вариант 2 - на основе синтаксиса on conflict и ограничения на Primary Key

insert into public.vip_users_channels(email, channel_id, channel_type)
values ('asd@asd.com', '111','teams')
on conflict on constraint vip_users_channels_pk do update set channel_id=<new_channel_id>;