CKAN：添加关联主库的审计元数据数据库并实现CKAN环境可见的问询

当然可行！ #

CKAN底层基于Flask和SQLAlchemy，天生支持多数据库连接配置，完全可以新增一个独立数据库来存储审计元数据，并且能和主数据库的表建立关联关系。下面给你具体的实现步骤和示例：

实现示例 #

1. 配置额外数据库连接 #

打开CKAN的配置文件（比如production.ini），在原有主数据库配置下方，新增审计数据库的连接信息：

# 主数据库配置保持不变
sqlalchemy.url = postgresql://ckanuser:your_password@localhost/ckan_main

# 新增审计数据库的URL
ckan.audit_db.url = postgresql://audit_user:audit_pass@localhost/ckan_audit

2. 创建审计元数据模型 #

在你的CKAN扩展里，新建模型文件，用SQLAlchemy定义审计表，注意指定绑定到审计数据库：

from sqlalchemy import Column, Integer, String, DateTime, ForeignKey
from sqlalchemy.ext.declarative import declarative_base
from ckan.model.meta import metadata

# 为审计库创建独立的Base类
AuditBase = declarative_base(metadata=metadata)

class AuditLog(AuditBase):
    __tablename__ = 'audit_log'
    __table_args__ = {'bind_key': 'audit'}  # 关键：指定使用审计数据库的绑定

    id = Column(Integer, primary_key=True, autoincrement=True)
    user_id = Column(String, ForeignKey('user.id'))  # 关联主数据库的user表
    action_type = Column(String(60), nullable=False)  # 比如resource_create、dataset_update
    action_time = Column(DateTime, nullable=False)
    target_id = Column(String)  # 操作的目标ID，比如资源ID、数据集ID

3. 初始化审计数据库表 #

在扩展的初始化脚本（比如plugin.py的IConfigurable实现里），添加创建表的逻辑：

from sqlalchemy import create_engine
from ckan.common import config
from .model import AuditBase

def init_audit_db():
    audit_db_url = config.get('ckan.audit_db.url')
    if not audit_db_url:
        return
    # 创建审计数据库的引擎
    audit_engine = create_engine(audit_db_url)
    # 创建所有审计表
    AuditBase.metadata.create_all(audit_engine)

# 在扩展的configure方法里调用初始化逻辑
class YourAuditPlugin(plugins.SingletonPlugin):
    plugins.implements(plugins.IConfigurable)

    def configure(self, config):
        init_audit_db()

4. 在业务逻辑中写入审计日志 #

比如在监听CKAN的动作事件（比如resource_created）时，写入审计记录：

from sqlalchemy.orm import sessionmaker
from ckan.common import config
from .model import AuditLog
from datetime import datetime

def write_audit_log(user_id, action_type, target_id):
    audit_db_url = config.get('ckan.audit_db.url')
    audit_engine = create_engine(audit_db_url)
    AuditSession = sessionmaker(bind=audit_engine)
    session = AuditSession()
    
    try:
        log_entry = AuditLog(
            user_id=user_id,
            action_type=action_type,
            action_time=datetime.utcnow(),
            target_id=target_id
        )
        session.add(log_entry)
        session.commit()
    except Exception as e:
        session.rollback()
        raise e
    finally:
        session.close()

# 监听CKAN事件并记录审计日志
class YourAuditPlugin(plugins.SingletonPlugin):
    plugins.implements(plugins.IActions)

    def get_actions(self):
        return {
            'resource_create': self.wrap_resource_create
        }

    def wrap_resource_create(self, context, data_dict):
        # 先执行原动作
        result = plugins.toolkit.get_action('resource_create')(context, data_dict)
        # 写入审计日志
        write_audit_log(
            user_id=context['user'],
            action_type='resource_create',
            target_id=result['id']
        )
        return result

替代方案（如果无法新增数据库） #

如果因为部署环境限制（比如没有权限创建新数据库），可以考虑这些替代方式：

• 主数据库新增审计表：直接在CKAN主数据库里添加审计相关的表，不需要额外数据库，实现最简单，适合小规模审计需求。
• 利用CKAN日志扩展：借助ckanext-logstash这类扩展，把审计日志发送到Elasticsearch、Logstash这类日志系统，无需数据库存储。
• 文件存储：对于低频率的审计记录，可以把日志序列化为JSON格式存在文件系统里，但只适合数据量较小的场景。

内容的提问来源于stack exchange，提问作者osum