Как активировать 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;

Как вызываются процедуры и функции на 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();
}