Postgres

• Как использовать процедуры и функции на Pl/SQL, какие есть еще языки для работы с Postgres?
• Как активировать PL/Python, примеры функции и процедуры
• Использование Postgres в C#
• Как вызываются процедуры и функции на PL/Python из C#?
• Как передавать и парсить JSON/JSONB из C# в PL/Python функцию через Npgsql?
• Какой тип даных поля в таблице Postgres лучше использовать для хранения пакетов json, но там могут быть и другие типы, например XML или изображения?
• Запрос данных из пакетов JSON в поле таблицы
• Работа с Postgres SQL
• Переключение текущей сессии в нужную схему.

Как использовать процедуры и функции на Pl/SQL, какие есть еще языки для работы с Postgres?

Функции (Functions)

CREATE OR REPLACE FUNCTION calculate_bonus(emp_id INT, bonus_percentage NUMERIC)
RETURNS NUMERIC AS $$
DECLARE
    current_salary NUMERIC;
    calculated_bonus NUMERIC;
BEGIN
    -- Получаем текущую зарплату сотрудника
    SELECT salary INTO current_salary 
    FROM employees 
    WHERE id = emp_id;
    
    -- Если сотрудник не найден, возвращаем 0
    IF current_salary IS NULL THEN
        RETURN 0;
    END IF;

    calculated_bonus := current_salary * (bonus_percentage / 100);
    RETURN calculated_bonus;
END;
$$ LANGUAGE plpgsql;

SELECT calculate_bonus(5, 10.5);

Процедуры (Procedures)

CREATE OR REPLACE PROCEDURE transfer_funds(
    sender_account INT, 
    receiver_account INT, 
    amount NUMERIC
)
LANGUAGE plpgsql
AS $$
BEGIN
    -- Снимаем средства с одного счета
    UPDATE accounts 
    SET balance = balance - amount 
    WHERE id = sender_account;

    -- Зачисляем средства на другой счет
    UPDATE accounts 
    SET balance = balance + amount 
    WHERE id = receiver_account;

    -- Фиксируем транзакцию
    COMMIT;
END;
$$;

CALL transfer_funds(1, 2, 500.00);

Основные управляющие конструкции PL/pgSQL

Чем различаются языки PL/SQL у Oracle и Postgres?

Основные различия

Примеры синтаксических отличий

Какие еще языки могут использоваться с postgres?

1. Языки из стандартного дистрибутива

2. Сторонние и популярные языки (расширения)

Когда и что использовать?

Как активировать PL/Python, примеры функции и процедуры

1. Активация PL/Python

CREATE EXTENSION IF NOT EXISTS plpython3u;

Важная деталь: Буква u на конце расширения означает untrusted (недоверенный). Это значит, что функции на PL/Python могут обращаться к файловой системе сервера, сети и операционной системе, поэтому создавать функции на PL/Python может только суперпользователь СУБД.

2. Пример функции (Function)

CREATE OR REPLACE FUNCTION calculate_cart_total(cart_json text)
RETURNS numeric AS $$
    import json

    try:
        data = json.loads(cart_json)
        items = data.get("items", [])
        
        total = 0.0
        for item in items:
            price = item.get("price", 0)
            quantity = item.get("quantity", 1)
            total += price * quantity
            
        return round(total, 2)
    except Exception as e:
        # raise_notice или log позволяют отправлять сообщения в клиент/лог базы
        plpy.notice(f"Ошибка при обработке JSON: {e}")
        return 0.0
$$ LANGUAGE plpython3u;

SELECT calculate_cart_total('{"items": [{"price": 100.5, "quantity": 2}, {"price": 50.0, "quantity": 1}]}');
-- Вернет: 251.00

3. Пример процедуры (Procedure)

CREATE OR REPLACE PROCEDURE cleanup_old_logs(days_old int)
AS $$
    import time

    # Выполняем SQL-запрос к БД с помощью plpy.execute
    plan = plpy.prepare(
        "SELECT id, message FROM system_logs WHERE created_at < NOW() - ($1 || ' days')::INTERVAL",
        ["int"]
    )
    result = plpy.execute(plan, [days_old])

    plpy.info(f"Найдено записей для удаления: {len(result)}")

    if len(result) > 0:
        # Логика обработки записей на Python
        for row in result:
            log_id = row["id"]
            # Здесь можно вызвать внешний API, сохранить резервную копию и т.д.

        # Удаление записей
        delete_plan = plpy.prepare(
            "DELETE FROM system_logs WHERE created_at < NOW() - ($1 || ' days')::INTERVAL",
            ["int"]
        )
        plpy.execute(delete_plan, [days_old])
        
        # Фиксация изменений
        plpy.commit()
$$ LANGUAGE plpython3u;

CALL cleanup_old_logs(30);

Дополнительные возможности plpy

Возвращают ли функции PL/Python наборы строк (recordsets) которые можно использовать в SQL-запросах?

Вариант 1. Возврат всех строк через список (List of Dicts)

CREATE OR REPLACE FUNCTION generate_user_report(min_age int)
RETURNS TABLE(user_id int, full_name text, status text) AS $$
    # В реальности данные могут запрашиваться из внешнего API или рассчитываться
    users = [
        {"user_id": 1, "full_name": "Алексей Иванов", "age": 25},
        {"user_id": 2, "full_name": "Мария Сидорова", "age": 19},
        {"user_id": 3, "full_name": "Иван Петров", "age": 34}
    ]
    
    result = []
    for user in users:
        if user["age"] >= min_age:
            result.append({
                "user_id": user["user_id"],
                "full_name": user["full_name"],
                "status": "Adult" if user["age"] >= 21 else "Young"
            })
            
    return result
$$ LANGUAGE plpython3u;

Использование функции в SQL:

SELECT * 
FROM generate_user_report(20)
WHERE status = 'Adult';

Вариант 2. Построчная отдача через генератор (yield)

CREATE OR REPLACE FUNCTION stream_numbers(limit_num int)
RETURNS TABLE(num int, square int) AS $$
    for i in range(1, limit_num + 1):
        # Отдаем по одной строке за раз
        yield {"num": i, "square": i * i}
$$ LANGUAGE plpython3u;

Использование в SQL-запросе:

SELECT num, square 
FROM stream_numbers(5) 
WHERE square > 10;

Вариант 3. Возврат результата встроенного SQL-запроса (plpy.execute)

CREATE OR REPLACE FUNCTION get_active_users()
RETURNS TABLE(id int, email text) AS $$
    # plpy.execute возвращает результат, который совместим с SETOF/TABLE
    res = plpy.execute("SELECT id, email FROM users WHERE is_active = true")
    return res
$$ LANGUAGE plpython3u;

Использование Postgres в C#

Как вызываются процедуры и функции на PL/Python из C#?

Вызов функции, возвращающей значение или набор строк

Пример 1: Вызов функции, возвращающей набор строк (RETURNS TABLE)

CREATE OR REPLACE FUNCTION generate_user_report(min_age int)
RETURNS TABLE(user_id int, full_name text, status text) AS $$
    # PL/Python логика
$$ LANGUAGE plpython3u;

using System;
using System.Threading.Tasks;
using Npgsql;

public class PostgresExample
{
    private const string ConnectionString = "Host=localhost;Database=mydb;Username=postgres;Password=secret";

    public async Task ReadUserReportAsync(int minAge)
    {
        await using var conn = new NpgsqlConnection(ConnectionString);
        await conn.OpenAsync();

        // Обращаемся к функции как к обычной таблице
        const string sql = "SELECT user_id, full_name, status FROM generate_user_report(@minAge)";

        await using var cmd = new NpgsqlCommand(sql, conn);
        cmd.Parameters.AddWithValue("minAge", minAge);

        await using var reader = await cmd.ExecuteReaderAsync();
        
        while (await reader.ReadAsync())
        {
            // Используем кортежи (Tuples), доступные начиная с C# 7
            var (userId, fullName, status) = (
                reader.GetInt32(0),
                reader.GetString(1),
                reader.GetString(2)
            );

            Console.WriteLine($"ID: {userId}, Имя: {fullName}, Статус: {status}");
        }
    }
}

Пример 2: Вызов скалярной функции (возвращает одно значение)

public async Task<decimal> GetCartTotalAsync(string cartJson)
{
    await using var conn = new NpgsqlConnection(ConnectionString);
    await conn.OpenAsync();

    const string sql = "SELECT calculate_cart_total(@cartJson)";

    await using var cmd = new NpgsqlCommand(sql, conn);
    cmd.Parameters.AddWithValue("cartJson", cartJson);

    // ExecuteScalarAsync возвращает первое значение первой строки
    var result = await cmd.ExecuteScalarAsync();
    return Convert.ToDecimal(result);
}

Вызов процедуры (Procedure)

Пример вызова процедуры через CALL

CREATE OR REPLACE PROCEDURE cleanup_old_logs(days_old int) AS $$
    # PL/Python логика
$$ LANGUAGE plpython3u;

public async Task CleanupLogsAsync(int daysOld)
{
    await using var conn = new NpgsqlConnection(ConnectionString);
    await conn.OpenAsync();

    // Явный вызов процедуры
    const string sql = "CALL cleanup_old_logs(@daysOld)";

    await using var cmd = new NpgsqlCommand(sql, conn);
    cmd.Parameters.AddWithValue("daysOld", daysOld);

    // Используем ExecuteNonQueryAsync, так как процедура не возвращает табличный результат
    await cmd.ExecuteNonQueryAsync();
}

Особенности работы с массивами (ARRAY) из C# 7

public async Task ProcessIdsAsync(int[] ids)
{
    await using var conn = new NpgsqlConnection(ConnectionString);
    await conn.OpenAsync();

    const string sql = "SELECT process_array_function(@ids)";

    await using var cmd = new NpgsqlCommand(sql, conn);
    // Npgsql автоматически приведёт int[] C# к integer[] в Postgres
    cmd.Parameters.AddWithValue("ids", ids);

    await cmd.ExecuteNonQueryAsync();
}

Как передавать и парсить JSON/JSONB из C# в PL/Python функцию через Npgsql?

1. Подготовка функции в PostgreSQL (PL/Python)

CREATE OR REPLACE FUNCTION process_user_data(payload jsonb)
RETURNS TABLE(user_id int, status text, is_valid boolean) AS $$
    import json

    # PL/Python получает json/jsonb как строку
    data = json.loads(payload)

    users = data.get("users", [])
    result = []

    for user in users:
        u_id = user.get("id")
        email = user.get("email", "")
        
        # Простая проверка/обработка
        is_valid = "@" in email
        status = "Active" if user.get("age", 0) >= 18 else "Pending"
        
        result.append({
            "user_id": u_id,
            "status": status,
            "is_valid": is_valid
        })

    return result
$$ LANGUAGE plpython3u;

2. Вызов из C# с помощью Npgsql и System.Text.Json

Вариант 1: Использование System.Text.Json (рекомендуемый)

using System;
using System.Collections.Generic;
using System.Text.Json;
using System.Threading.Tasks;
using Npgsql;
using NpgsqlTypes;

public class JsonPostgresExample
{
    private const string ConnectionString = "Host=localhost;Database=mydb;Username=postgres;Password=secret";

    public async Task ProcessDataAsync()
    {
        // 1. Формируем C# объект для передачи
        var requestData = new
        {
            users = new[]
            {
                new { id = 101, email = "alex@example.com", age = 25 },
                new { id = 102, email = "invalid_email", age = 16 }
            }
        };

        // 2. Сериализуем объект в строку JSON
        string jsonPayload = JsonSerializer.Serialize(requestData);

        await using var conn = new NpgsqlConnection(ConnectionString);
        await conn.OpenAsync();

        const string sql = "SELECT user_id, status, is_valid FROM process_user_data(@payload)";

        await using var cmd = new NpgsqlCommand(sql, conn);

        // 3. Добавляем параметр с явным указанием NpgsqlDbType.Jsonb
        cmd.Parameters.Add(new NpgsqlParameter("payload", NpgsqlDbType.Jsonb) 
        { 
            Value = jsonPayload 
        });

        // 4. Читаем результат
        await using var reader = await cmd.ExecuteReaderAsync();
        while (await reader.ReadAsync())
        {
            var userId = reader.GetInt32(0);
            var status = reader.GetString(1);
            var isValid = reader.GetBoolean(2);

            Console.WriteLine($"ID: {userId} | Статус: {status} | Валиден: {isValid}");
        }
    }
}

3. Обратный сценарий: Возврат JSONB из PL/Python в C#

Функция в PostgreSQL:

CREATE OR REPLACE FUNCTION get_system_stats()
RETURNS jsonb AS $$
    import json

    stats = {
        "server_time": "2026-09-12",
        "metrics": {
            "cpu_usage": 42.5,
            "active_connections": 12
        },
        "tags": ["prod", "primary"]
    }

    # Возвращаем JSON-строку
    return json.dumps(stats)
$$ LANGUAGE plpython3u;

Код чтения в C#:

public async Task GetStatsAsync()
{
    await using var conn = new NpgsqlConnection(ConnectionString);
    await conn.OpenAsync();

    const string sql = "SELECT get_system_stats()";
    await using var cmd = new NpgsqlCommand(sql, conn);

    var rawJson = (string)await cmd.ExecuteScalarAsync();

    // Десериализуем JSON в C# объект
    using var doc = JsonDocument.Parse(rawJson);
    var cpu = doc.RootElement
                 .GetProperty("metrics")
                 .GetProperty("cpu_usage")
                 .GetDouble();

    Console.WriteLine($"CPU Usage: {cpu}%");
}

Ключевые моменты при работе с JSON/JSONB:

 

Какой тип даных поля в таблице Postgres лучше использовать для хранения пакетов json, но там могут быть и другие типы, например XML или изображения?

Подход 1: BYTEA (Бинарные данные) — Самый универсальный и производительный

CREATE TABLE payload_store (
    id BIGINT GENERATED ALWAYS AS IDENTITY PRIMARY KEY,
    content_type VARCHAR(50) NOT NULL, -- 'application/json', 'application/xml', 'image/png'
    payload BYTEA NOT NULL,
    created_at TIMESTAMPTZ DEFAULT NOW()
);

Подход 2: TEXT — Удобнее для отладки, но со штрафом для бинарных данных

CREATE TABLE payload_store (
    id BIGINT GENERATED ALWAYS AS IDENTITY PRIMARY KEY,
    content_type VARCHAR(50) NOT NULL,
    payload TEXT NOT NULL
);

Сравнение вариантов

Рекомендация

Как работать с BYTEA из C# (Npgsql):

// Для JSON/XML кодируем строку в UTF-8 байты:
byte[] jsonBytes = System.Text.Encoding.UTF8.GetBytes(jsonString);

// Для изображений берем байты файла напрямую:
byte[] imageBytes = await System.IO.File.ReadAllBytesAsync("photo.png");

// Передача в NpgsqlParameter:
cmd.Parameters.Add(new NpgsqlParameter("payload", NpgsqlTypes.NpgsqlDbType.Bytea) { Value = imageBytes });

Запрос данных из пакетов JSON в поле таблицы

Для хранения пакетов используем такую таблицу:

Пакеты хранятся в поле pack_data, они одинаковой структуры для определенного типа записи (поле type_id)

Пакет имеет такую структуру:

[
  {"charsetinfo": 
    {"name": "utf-8"}
  },
  {"history": 
    [
      {"BOARDID": "TQBR", "TRADEDATE": "2026-09-21", "SHORTNAME": "РоссЮг ао", "SECID": "MRKY", "NUMTRADES": 275, "VALUE": 1541379, "OPEN": 0.07065, "LOW": 0.0695, "HIGH": 0.0716, "LEGALCLOSEPRICE": 0.07015, "WAPRICE": 0.0704, "CLOSE": 0.07055, "VOLUME": 21890000, "MARKETPRICE2": 0.0702, "MARKETPRICE3": 0.0702, "ADMITTEDQUOTE": null, "MP2VALTRD": 766668, "MARKETPRICE3TRADESVALUE": 766668, "ADMITTEDVALUE": null, "WAVAL": 0, "TRADINGSESSION": 3, "CURRENCYID": "SUR", "TRENDCLSPR": 0.21, "TRADE_SESSION_DATE": "2026-09-21"},{"BOARDID": "TQBR", "TRADEDATE": "2026-09-21", "SHORTNAME": "Мостотрест", "SECID": "MSTT", "NUMTRADES": 162, "VALUE": 1539552, "OPEN": 69.4, "LOW": 67.9, "HIGH": 69.4, "LEGALCLOSEPRICE": 67.9, "WAPRICE": 68.5, "CLOSE": 68.35, "VOLUME": 22470, "MARKETPRICE2": 68.45, "MARKETPRICE3": 68.45, "ADMITTEDQUOTE": null, "MP2VALTRD": 1292162.5, "MARKETPRICE3TRADESVALUE": 1292162.5, "ADMITTEDVALUE": null, "WAVAL": 0, "TRADINGSESSION": 3, "CURRENCYID": "SUR", "TRENDCLSPR": -1.51, "TRADE_SESSION_DATE": "2026-09-21"},
        ...
     ]
  }
]

Нам нужно соединить все пакеты и получить записи из поля "history".

Чтобы развернуть массив из массива history во всех JSON-пакетах одной SQL-выборкой, в PostgreSQL используются функции для работы с JSON/JSONB:

SQL-запрос

Поскольку yаш JSON представляет собой массив объектов, где массив history находится во втором элементе (pack_data::jsonb -> 1 -> 'history'), запрос будет выглядеть так:

SELECT u.upload_id,
       u.pack_name,
       u.upload_time,
       -- Извлекаем поля из объектов внутри history
       history_item ->> 'SECID' AS secid,
       history_item ->> 'SHORTNAME' AS shortname,
       history_item ->> 'TRADEDATE' AS tradedate,
       (history_item ->> 'OPEN')::numeric AS open_price,
       (history_item ->> 'CLOSE')::numeric AS close_price,
       (history_item ->> 'VOLUME')::bigint AS volume,
       (history_item ->> 'VALUE')::numeric AS value,
       history_item ->> 'BOARDID' AS boardid
  FROM trading.upload_data u
 WHERE u.type_id = 1
 -- 1. Извлекаем массив history из второго элемента корня JSON и разворачиваем его по строкам
 CROSS JOIN LATERAL jsonb_array_elements(u.pack_data::jsonb -> 1 -> 'history') AS history_item;

Как это работает:

Альтернатива: Если порядок элементов в корневом JSON может меняться

Если элемент history не всегда гарантированно идет вторым (с индексом 1), безопаснее сначала распаковать корневой массив, а затем найти элемент с ключом history:

SELECT u.upload_id,
       history_item ->> 'SECID' AS secid,
       history_item ->> 'SHORTNAME' AS shortname,
       (history_item ->> 'CLOSE')::numeric AS close_price
  FROM trading.upload_data u
 WHERE u.type_id = 1
 -- Распаковываем корневой массив
 CROSS JOIN LATERAL jsonb_array_elements(u.pack_data::jsonb) AS root_elem
 -- Распаковываем массив history (игнорируя элементы без этого ключа, например charsetinfo)
 CROSS JOIN LATERAL jsonb_array_elements(root_elem -> 'history') AS history_item;

Этот вариант гибче, так как найдет массив history в любом месте структуры пакета.

Что делает оператор:  ->>

Примеры применения

1. Извлечение поля по ключу

-- Оператор -> возвращает JSON ("Иван" в кавычках)
SELECT data -> 'user' -> 'name'; 
-- Результат: "Иван" (тип jsonb)

-- Оператор ->> возвращает чистый текст (без кавычек)
SELECT data -> 'user' ->> 'name'; 
-- Результат: Иван (тип text)

2. Извлечение элемента из JSON-массива по индексу

SELECT data ->> 0; 
-- Результат: apple (тип text)

3. Приведение типов для расчетов и фильтрации

-- Извлекаем возраст как число для условия в WHERE
SELECT * 
FROM users_table 
WHERE (data ->> 'age')::int >= 18;

-- Извлекаем дату и делаем сортировку
SELECT data ->> 'TRADEDATE' AS trade_date
FROM trading.upload_data
ORDER BY (data ->> 'TRADEDATE')::date DESC;

Чек-лист для запоминания

Работа с Postgres SQL

Переключение текущей сессии в нужную схему.

Допустим вам нужно сделать схему main текущей.

Это достигается выполнением следующей команды:

SET search_path TO main, public;

Как это работает: Любые команды CREATE TABLE qrtz_... без явного указания схемы теперь будут создавать таблицы строго внутри main.

Проверьте переключение

Чтобы убедиться, что контекст изменился на main, выполните:

SHOW search_path;
SELECT current_schema();

В ответе должно быть main.

Что будет, если указать только main?