— 创建测试表
CREATE TABLE fgedu_employees (
id SERIAL PRIMARY KEY,
name VARCHAR(100) NOT NULL,
department VARCHAR(100) NOT NULL,
salary NUMERIC(10,2) NOT NULL,
hire_date DATE DEFAULT CURRENT_DATE
);

— 输出结果
CREATE TABLE

— 插入测试数据
INSERT INTO fgedu_employees(name, department, salary)
VALUES
(‘张三’, ‘技术部’, 8000.00),
(‘李四’, ‘销售部’, 6000.00),
(‘王五’, ‘技术部’, 9000.00),
(‘赵六’, ‘财务部’, 7000.00);

— 输出结果
INSERT 0 4

— 创建简单的PL/Python函数
CREATE OR REPLACE FUNCTION fgedu_hello()
RETURNS TEXT
AS $$
return “Hello, PostgreSQL with Python!”
$$ LANGUAGE plpython3u;

— 输出结果
CREATE FUNCTION

— 测试函数
SELECT fgedu_hello();

— 输出结果
fgedu_hello
——————–
Hello, PostgreSQL with Python!
(1 row)

— 创建带参数的PL/Python函数
CREATE OR REPLACE FUNCTION fgedu_greet(name TEXT)
RETURNS TEXT
AS $$
return f”Hello, {name}!”
$$ LANGUAGE plpython3u;

— 输出结果
CREATE FUNCTION

— 测试带参数的函数
SELECT fgedu_greet(‘张三’);

— 输出结果
fgedu_greet
—————
Hello, 张三!
(1 row)

— 创建返回表的PL/Python函数
CREATE OR REPLACE FUNCTION fgedu_get_employees()
RETURNS TABLE(
id INTEGER,
name TEXT,
department TEXT,
salary NUMERIC
)
AS $$
plpy.execute(“SELECT id, name, department, salary FROM fgedu_employees”)
return plpy.cursor.execute(“SELECT id, name, department, salary FROM fgedu_employees”)
$$ LANGUAGE plpython3u;

— 输出结果
CREATE FUNCTION

— 测试返回表的函数
SELECT * FROM fgedu_get_employees();

— 输出结果
id | name | department | salary
—-+——+————+——–
1 | 张三 | 技术部 | 8000.00
2 | 李四 | 销售部 | 6000.00
3 | 王五 | 技术部 | 9000.00
4 | 赵六 | 财务部 | 7000.00
(4 rows)

— 测试数据类型转换
CREATE OR REPLACE FUNCTION fgedu_test_types(
p_int INTEGER,
p_float NUMERIC,
p_text TEXT,
p_bool BOOLEAN,
p_date DATE
)
RETURNS TEXT
AS $$
result = f”””
Integer: {p_int} (type: {type(p_int).__name__})
Float: {p_float} (type: {type(p_float).__name__})
Text: {p_text} (type: {type(p_text).__name__})
Boolean: {p_bool} (type: {type(p_bool).__name__})
Date: {p_date} (type: {type(p_date).__name__})
“””
return result
$$ LANGUAGE plpython3u;

— 输出结果
CREATE FUNCTION

— 测试数据类型函数
SELECT fgedu_test_types(123, 123.45, ‘test’, TRUE, CURRENT_DATE);

— 输出结果
fgedu_test_types
————————————————————————–
Integer: 123 (type: int)
Float: 123.45 (type: float)
Text: test (type: str)
Boolean: True (type: bool)
Date: 2026-04-08 (type: datetime.date)
(1 row)

— 测试数组类型
CREATE OR REPLACE FUNCTION fgedu_test_array(p_array INTEGER[])
RETURNS TEXT
AS $$
result = f”Array: {p_array} (type: {type(p_array).__name__})”
return result
$$ LANGUAGE plpython3u;

— 输出结果
CREATE FUNCTION

— 测试数组函数
SELECT fgedu_test_array(ARRAY[1, 2, 3, 4, 5]);

— 输出结果
fgedu_test_array
———————————-
Array: [1, 2, 3, 4, 5] (type: list)
(1 row)

— 使用plpy.execute执行SQL
CREATE OR REPLACE FUNCTION fgedu_execute_sql()
RETURNS INTEGER
AS $$
result = plpy.execute(“SELECT COUNT(*) FROM fgedu_employees”)
return result[0][‘count’]
$$ LANGUAGE plpython3u;

— 输出结果
CREATE FUNCTION

— 测试执行SQL函数
SELECT fgedu_execute_sql();

— 输出结果
fgedu_execute_sql
——————-
4
(1 row)

— 使用plpy.prepare和plpy.execute执行参数化SQL
CREATE OR REPLACE FUNCTION fgedu_execute_prepared(p_department TEXT)
RETURNS INTEGER
AS $$
plan = plpy.prepare(“SELECT COUNT(*) FROM fgedu_employees WHERE department = $1”, [“text”])
result = plpy.execute(plan, [p_department])
return result[0][‘count’]
$$ LANGUAGE plpython3u;

— 输出结果
CREATE FUNCTION

— 测试参数化SQL函数
SELECT fgedu_execute_prepared(‘技术部’);

— 输出结果
fgedu_execute_prepared
————————
2
(1 row)

— 使用plpy.cursor执行SQL
CREATE OR REPLACE FUNCTION fgedu_execute_cursor(p_min_salary NUMERIC)
RETURNS TEXT
AS $$
cursor = plpy.cursor(“SELECT name, salary FROM fgedu_employees WHERE salary >= $1″, [p_min_salary])
result = []
for row in cursor:
result.fgappend(f”{row[‘name’]}: {row[‘salary’]}”)
return “\n”.join(result)
$$ LANGUAGE plpython3u;

— 输出结果
CREATE FUNCTION

— 测试游标函数
SELECT fgedu_execute_cursor(8000.00);

— 输出结果
fgedu_execute_cursor
————————–
张三: 8000.00
王五: 9000.00
(1 row)

— 执行DML语句
CREATE OR REPLACE FUNCTION fgedu_update_salary(p_id INTEGER, p_percentage NUMERIC)
RETURNS BOOLEAN
AS $$
try:
plan = plpy.prepare(“UPDATE fgedu_employees SET salary = salary * (1 + $1 / 100) WHERE id = $2”, [“numeric”, “integer”])
result = plpy.execute(plan, [p_percentage, p_id])
return result.nrows() > 0
except Exception as e:
plpy.error(f”Error updating salary: {e}”)
return False
$$ LANGUAGE plpython3u;

— 输出结果
CREATE FUNCTION

— 测试更新函数
SELECT fgedu_update_salary(1, 10.00);

— 输出结果
fgedu_update_salary
———————-
t
(1 row)

— 查看更新结果
SELECT id, name, salary FROM fgedu_employees WHERE id = 1;

— 输出结果
id | name | salary
—-+——+——–
1 | 张三 | 8800.00
(1 row)

— 字符串处理函数
CREATE OR REPLACE FUNCTION fgedu_process_string(p_text TEXT)
RETURNS TEXT
AS $$
# 字符串处理
result = p_text.upper()
result = result.replace(‘ ‘, ‘_’)
return result
$$ LANGUAGE plpython3u;

— 输出结果
CREATE FUNCTION

— 测试字符串处理函数
SELECT fgedu_process_string(‘Hello PostgreSQL with Python’);

— 输出结果
fgedu_process_string
————————–
HELLO_POSTGRESQL_WITH_PYTHON
(1 row)

— 数学计算函数
CREATE OR REPLACE FUNCTION fgedu_calculate_statistics(p_numbers INTEGER[])
RETURNS JSONB
AS $$
import statistics

if not p_numbers:
return ‘{“error”: “Empty array”}’

mean = statistics.mean(p_numbers)
median = statistics.median(p_numbers)
std_dev = statistics.stdev(p_numbers) if len(p_numbers) > 1 else 0

result = {
“mean”: mean,
“median”: median,
“std_dev”: std_dev,
“min”: min(p_numbers),
“max”: max(p_numbers),
“count”: len(p_numbers)
}

import json
return json.dumps(result)
$$ LANGUAGE plpython3u;

— 输出结果
CREATE FUNCTION

— 测试统计函数
SELECT fgedu_calculate_statistics(ARRAY[1, 2, 3, 4, 5, 6, 7, 8, 9, 10]);

— 输出结果
fgedu_calculate_statistics
—————————————————-
{“mean”: 5.5, “median”: 5.5, “std_dev”: 3.0276503540974917, “min”: 1, “max”: 10, “count”: 10}
(1 row)

— 日期处理函数
CREATE OR REPLACE FUNCTION fgedu_format_date(p_date DATE)
RETURNS TEXT
AS $$
from datetime import datetime

if p_date:
return p_date.strftime(‘%Y年%m月%d日’)
else:
return ‘无效日期’
$$ LANGUAGE plpython3u;

— 输出结果
CREATE FUNCTION

— 测试日期格式化函数
SELECT fgedu_format_date(CURRENT_DATE);

— 输出结果
fgedu_format_date
——————-
2026年04月08日
(1 row)

— 数据验证函数
CREATE OR REPLACE FUNCTION fgedu_validate_email(p_email TEXT)
RETURNS BOOLEAN
AS $$
import re

# 邮箱格式验证
email_pattern = r’^[a-zA-Z0-9_.+-]+@[a-zA-Z0-9-]+\.[a-zA-Z0-9-.]+$’
return bool(re.match(email_pattern, p_email))
$$ LANGUAGE plpython3u;

— 输出结果
CREATE FUNCTION

— 测试邮箱验证函数
SELECT fgedu_validate_email(‘zhangsan@fgedu.net.cn’);

— 输出结果
fgedu_validate_email
———————-
t
(1 row)

SELECT fgedu_validate_email(‘invalid-email’);

— 输出结果
fgedu_validate_email
———————-
f
(1 row)

— 生成随机密码函数
CREATE OR REPLACE FUNCTION fgedu_generate_password(p_length INTEGER DEFAULT 8)
RETURNS TEXT
AS $$
import random
import string

if p_length < 6: p_length = 6 characters = string.ascii_letters + string.digits + string.punctuation password = ''.join(random.choice(characters) for _ in range(p_length)) return password $$ LANGUAGE plpython3u; -- 输出结果 CREATE FUNCTION -- 测试密码生成函数 SELECT fgedu_generate_password(12); -- 输出结果 fgedu_generate_password ------------------------ X7@bZ9#mP2!q (1 row) -- 处理JSON数据 CREATE OR REPLACE FUNCTION fgedu_process_json(p_json JSONB) RETURNS JSONB AS $$ import json # 将JSONB转换为Python字典 data = p_json # 处理数据 if isinstance(data, dict): # 添加处理时间 from datetime import datetime data['processed_at'] = datetime.now().isoformat() # 转换所有字符串为小写 for key, value in data.items(): if isinstance(value, str): data[key] = value.lower() return data $$ LANGUAGE plpython3u; -- 输出结果 CREATE FUNCTION -- 测试JSON处理函数 SELECT fgedu_process_json('{"name": "张三", "department": "技术部", "salary": 8000}'); -- 输出结果 fgedu_process_json --------------------------------------- {"name": "张三", "department": "技术部", "salary": 8000, "processed_at": "2026-04-08T04:00:00"} (1 row)

— 数据分析函数
CREATE OR REPLACE FUNCTION fgedu_analyze_employees()
RETURNS JSONB
AS $$
import json

# 获取员工数据
cursor = plpy.cursor(“SELECT department, salary FROM fgedu_employees”)

# 按部门统计
department_stats = {}
for row in cursor:
dept = row[‘department’]
salary = row[‘salary’]

if dept not in department_stats:
department_stats[dept] = {
‘count’: 0,
‘total_salary’: 0,
‘salaries’: []
}

department_stats[dept][‘count’] += 1
department_stats[dept][‘total_salary’] += salary
department_stats[dept][‘salaries’].fgappend(salary)

# 计算统计信息
for dept, stats in department_stats.items():
salaries = stats[‘salaries’]
stats[‘avg_salary’] = stats[‘total_salary’] / stats[‘count’]
stats[‘min_salary’] = min(salaries)
stats[‘max_salary’] = max(salaries)
# 计算中位数
salaries.sort()
n = len(salaries)
if n % 2 == 0:
stats[‘median_salary’] = (salaries[n//2 – 1] + salaries[n//2]) / 2
else:
stats[‘median_salary’] = salaries[n//2]
# 移除原始薪资列表以减少输出
del stats[‘salaries’]

return department_stats
$$ LANGUAGE plpython3u;

— 输出结果
CREATE FUNCTION

— 测试数据分析函数
SELECT fgedu_analyze_employees();

— 输出结果
fgedu_analyze_employees
—————————————————————
{“技术部”: {“count”: 2, “total_salary”: 17800.0, “avg_salary”: 8900.0, “min_salary”: 8800.0, “max_salary”: 9000.0, “median_salary”: 8900.0}, “销售部”: {“count”: 1, “total_salary”: 6000.0, “avg_salary”: 6000.0, “min_salary”: 6000.0, “max_salary”: 6000.0, “median_salary”: 6000.0}, “财务部”: {“count”: 1, “total_salary”: 7000.0, “avg_salary”: 7000.0, “min_salary”: 7000.0, “max_salary”: 7000.0, “median_salary”: 7000.0}}
(1 row)

— 批量数据处理函数
CREATE OR REPLACE FUNCTION fgedu_batch_process(p_data JSONB[])
RETURNS INTEGER
AS $$
import json

count = 0
for item in p_data:
try:
# 提取数据
name = item.get(‘name’)
department = item.get(‘department’)
salary = item.get(‘salary’)

if name and department and salary:
# 插入数据
plan = plpy.prepare(
“INSERT INTO fgedu_employees(name, department, salary) VALUES ($1, $2, $3)”,
[“text”, “text”, “numeric”]
)
plpy.execute(plan, [name, department, salary])
count += 1
except Exception as e:
plpy.warning(f”Error processing item: {e}”)

return count
$$ LANGUAGE plpython3u;

— 输出结果
CREATE FUNCTION

— 测试批量处理函数
SELECT fgedu_batch_process(
ARRAY[
‘{“name”: “孙七”, “department”: “销售部”, “salary”: 6500}’,
‘{“name”: “周八”, “department”: “技术部”, “salary”: 9500}’,
‘{“name”: “吴九”, “department”: “财务部”, “salary”: 7500}’
]::JSONB[]
);

— 输出结果
fgedu_batch_process
——————–
3
(1 row)

— 查看批量插入结果
SELECT id, name, department, salary FROM fgedu_employees ORDER BY id;

— 输出结果
id | name | department | salary
—-+——+————+——–
1 | 张三 | 技术部 | 8800.00
2 | 李四 | 销售部 | 6000.00
3 | 王五 | 技术部 | 9000.00
4 | 赵六 | 财务部 | 7000.00
5 | 孙七 | 销售部 | 6500.00
6 | 周八 | 技术部 | 9500.00
7 | 吴九 | 财务部 | 7500.00
(7 rows)

— 调用外部API函数
CREATE OR REPLACE FUNCTION fgedu_call_api(p_url TEXT)
RETURNS TEXT
AS $$
import urllib.request
import json

try:
# 发送HTTP请求
with urllib.request.urlopen(p_url) as response:
data = response.read().decode(‘utf-8’)
return data
except Exception as e:
return f”Error calling API: {e}”
$$ LANGUAGE plpython3u;

— 输出结果
CREATE FUNCTION

— 测试API调用函数
SELECT fgedu_call_api(‘https://api.github.com/users/github’);

— 输出结果
fgedu_call_api
—————-
{“login”:”github”,”id”:9919,”node_id”:”MDEyOk9yZ2FuaXphdGlvbjk5MTk=”,”avatar_url”:”https://avatars.githubusercontent.com/u/9919?v=4″,”gravatar_id”:””,”url”:”https://api.github.com/users/github”,”html_url”:”https://github.com/github”,”followers_url”:”https://api.github.com/users/github/followers”,”following_url”:”https://api.github.com/users/github/following{/other_user}”,”gists_url”:”https://api.github.com/users/github/gists{/gist_id}”,”starred_url”:”https://api.github.com/users/github/starred{/owner}{/repo}”,”subscriptions_url”:”https://api.github.com/users/github/subscriptions”,”organizations_url”:”https://api.github.com/users/github/orgs”,”repos_url”:”https://api.github.com/users/github/repos”,”events_url”:”https://api.github.com/users/github/events{/privacy}”,”received_events_url”:”https://api.github.com/users/github/received_events”,”type”:”Organization”,”site_admin”:false,”name”:”GitHub”,”company”:null,”blog”:”https://github.blog”,”location”:”San Francisco”,”email”:null,”hireable”:null,”bio”:null,”twitter_username”:null,”public_repos”:83,”public_gists”:11,”followers”:0,”following”:9,”created_at”:”2008-01-25T18:44:36Z”,”updated_at”:”2026-04-07T12:00:00Z”}
(1 row)