ckanext-collection

CKAN中查看数据系列的基本类。

内容

• 需求
• 安装
• 使用
• 文档
  • 概述
  • 收集初始化
  • 服务
    • 通用逻辑
    • 数据服务
    • 分页服务
    • 序列化服务
    • 列服务
    • 过滤器服务
  • 核心类和用法示例
    • 收集
    • DbCollection
    • 数据
    • StaticData
    • BaseSaData
    • StatementSaData
    • UnionSaData
    • ModelData
    • TableData
    • ApiData
    • ApiSearchData
    • ApiListData
    • 分页器
    • ClassicPager
    • 列
    • TableColumns
    • 过滤器
    • 序列化器
    • CsvSerializer
    • JsonlSerializer
    • JsonSerializer
    • HtmlSerializer
    • TableSerializer
    • HtmxTableSerializer
• 配置设置
• 集成
  • ckanext-admin-panel
• 许可证

需求

与核心CKAN版本的兼容性

CKAN版本	兼容？
2.9	不兼容
2.10	是
主控	是

安装

安装 ckanext-collection

1. 安装扩展

   pip install ckanext-collection
   

2. 将 collection 添加到您的 CKAN 配置文件中的 ckan.plugins 设置。

使用

可以通过 ckanext.collection.interfaces.ICollection 或通过 CKAN 信号来注册集合。已注册的集合可以在代码的任何地方使用辅助函数进行初始化，并可用于多个通用端点，以将集合渲染为 HTML 或导出为不同的格式。

通过接口注册

from ckanext.collection.interfaces import CollectionFactory, ICollection


class MyPlugin(p.SingletonPlugin):
    p.implements(ICollection, inherit=True)

    def get_collection_factories(self) -> dict[str, CollectionFactory]:
        return {
            "my-collection": MyCollection,
        }

get_collection_factories 返回一个字典，键为集合名称（允许字母、数字、下划线和连字符），值为集合工厂。在大多数通用情况下，集合工厂只是一个集合类。但您可以使用任何具有签名 (str, dict[str, Any], **Any) -> Collection 的函数作为工厂。例如，以下函数是一个有效的集合工厂，并且可以从 get_collection_factories 返回

def my_factory(name: str, params: dict[str, Any], **kwargs: Any):
    """Collection that shows 100 numbers per page"""
    params.setdefault("rows_per_page", 100)
    return MyCollection(name, params, **kwargs)

如果您只想在集合插件启用时注册集合，则可以使用 CKAN 信号而不是将 ckanext-collection 的导入封装在 try except 块中

class MyPlugin(p.SingletonPlugin):
    p.implements(p.ISignal)

    def get_signal_subscriptions(self) -> types.SignalMapping:
        return {
            tk.signals.ckanext.signal("collection:register_collections"): [
                self.collect_collection_factories,
            ],
        }

    def collect_collection_factories(self, sender: None):
        return {
            "my-collection": MyCollection,
        }

从信号订阅返回的数据与从 ICollection.get_collection_factories 返回的数据完全相同。唯一的区别是，由于信号的内部实现，信号订阅接受 sender 参数，该参数始终为 None。

文档

概述

此插件的目标是为您提供用于处理数据集合的通用类。因此，它开箱即用功能不多，您需要编写一些代码才能看到结果。

大多数有用的类都位于 ckanext.collection.utils 模块中，以下所有示例都需要在脚本开头添加以下行：from ckanext.collection.utils import *。

让我们从基础知识开始。 ckanext-collection 为不同的用途定义了一些集合。最基本的集合是 Collection，但它没有定制就没有价值，所以我们将从 StaticCollection 开始。

col = StaticCollection("name", {})

任何集合的构造函数都有两个必填参数：名称和参数。名称主要用于内部，由任何组合的字母、数字、连字符和下划线组成。参数作为字典传递，并改变集合的内容。

在最基本的情况下，集合代表一组类似的项目：数据集、用户、组织、字典、数字等。因此，它可以转换为列表或迭代。

list(col)

for item in col:
    print(item)

我们的测试集合目前为空，所以您现在什么也看不到。通常，StaticCollection 包含在创建集合时指定的静态数据。但由于我们没有指定任何数据，集合为空。

要解决这个问题，我们必须配置负责数据生成的集合的一部分，使用其 设置。集合将其内部逻辑分成多个可配置的 服务，我们需要的服务称为 数据 服务。要修改它，我们可以将名为 data_settings 的命名参数传递给集合的构造函数

col = StaticCollection(
    "name", {},
    data_settings={"data": [1,2,3]}
)

现在再次迭代集合，您现在会看到结果。

for item in col:
    print(item)

这并不令人印象深刻，但您对 静态 集合的期望并不高，对吧？还有其他更智能的集合，但我们需要学习更多此扩展的概念才能使用它们，所以现在我们只简要看看它们。

注意：在数据量方面，集合有一定的限制。默认情况下，您将只能看到大约 10 条记录，即使您有更多。对于 StaticCollection 也是如此 - 如果您将数据服务的 data 属性设置为 range(1, 100)，您可以看到它。我们将在以后学习如何控制这些限制。

StaticCollection 与静态数据一起工作。它可以用于测试或作为尚未实现的集合的占位符。在罕见情况下，它可以与任意可迭代对象一起使用，以创建数据交互的标准接口。

ModelCollection 与 SQLAlchemy 模型一起工作。我们将使用其数据服务中的两个属性：model 和 is_scalar。前者设置集合处理的实际模型，而后者控制我们如何处理每个单独的记录。默认情况下，ModelCollection 以多个列的形式返回每个记录，但我们将设置 is_scalar=True 并为每个记录接收模型实例

col = ModelCollection(
    "", {},
    data_settings={"is_scalar": True, "model": model.User}
)

for user in col:
  assert isinstance(user, model.User)
  print(f"{user.name}, {user.email}")

ApiSearchCollection 与类似于 package_search 的 API 动作一起工作。它们必须使用 rows 和 start 参数进行分页，并且其结果必须包含 count 和 results 键。其数据服务接受名为 API 动作产生数据的 action 属性

col = ApiSearchCollection(
    "", {},
    data_settings={"action": "package_search"}
)

for pkg in col:
  print(f"{pkg['id']}: {pkg['title']}")

ApiListCollection 与类似于 package_list 的 API 动作一起工作。它们必须使用 limit 和 offset 参数进行分页，并且其结果必须表示为列表。

col = ApiListCollection(
    "", {},
    data_settings={"action": "package_list"}
)

for name in col:
  print(name)

ApiCollection 与类似于 user_list 的 API 动作一起工作。它们必须以列表的形式一次性返回所有记录。

col = ApiCollection(
    "", {},
    data_settings={"action": "user_list"}
)

for user in col:
  print(user["name"])

收集初始化

集合构造函数有两个必需参数：名称和参数。

名称用作集合标识符，最好在整个集合中保持此值的唯一性。例如，名称用于在将集合序列化为 HTML 表格时计算 id 属性。如果您渲染具有相同名称的两个集合，则页面上将出现两个相同的 ID。

参数通常由数据和分页服务用于搜索、排序等。集合不会保留所有参数。相反，它只存储具有以 <name>: 前缀键的项。即，如果集合具有名称 hello，并且您传递 {"hello:a": 1, "b": 2, "world:c": 3}，则集合将删除 b（因为它没有集合名称加冒号前缀）和 world:c 成员（因为它使用 world 而不是 hello 在前缀中）。至于 hello:a，集合从其中删除 <name>: 前缀。因此，最终集合存储 {"a": 1}。您可以使用 params 属性检查集合的参数

col = Collection("hello", {"hello:a": 1, "b": 2, "world:c": 3})
assert col.params == {"a": 1}

col = Collection("world", {"hello:a": 1, "b": 2, "world:c": 3})
assert col.params == {"c": 3}

它允许您在同一页上同时渲染和处理多个集合。想象一下，您有集合 users 和集合 packages。您想看到 users 的第二页和 packages 的第五页。提交查询字符串 ?users:page=2&packages:page=5 并使用以下代码初始化集合

from ckan.logic import parse_params
from ckan.plugins import toolkit as tk

params = parse_params(tk.request.args)

users = ModelCollection(
    "users", params,
    data_settings={"model": model.User}
)
packages = ModelCollection(
    "packages", params,
    data_settings={"model": model.Package}
)

assert users.pager.page == 2
assert packages.pager.page == 5

服务

集合本身仅包含最少的逻辑，而所有繁重的工作都委托给 服务。集合知道如何初始化服务，并且通常您所有集合之间唯一的区别是它们所有服务的配置方式。

集合包含以下服务

• data：控制可以从集合接收的确切数据。包含搜索、筛选、排序等逻辑。
• pager：定义数据迭代的限制。正是这个服务在迭代静态集合时只显示 10 条记录。
• serializer：指定如何将集合转换为所需的形式。使用正确的序列化器，您将能够将整个集合导出为 CSV、JSON、YAML 或将其渲染为 HTML 表格。
• columns：包含用于其他服务的特定数据列的配置。它可能定义用于导出到 CSV 的模型属性，应用于特定属性的转换函数的名称，以及可用于在数据的 HTML 表示形式中进行排序的列的名称。
• filters：包含在数据序列化过程中生成的额外小部件的配置。例如，当数据序列化为 HTML 表格时，筛选器可以定义下拉框和数据搜索表单的输入字段的配置。

注意：您可以在自定义集合中定义更多服务。上面的列表列出了基本集合以及当前扩展中提供的所有集合中可用的服务。例如，内置集合之一DbCollection有一个名为db_connection的附加服务，可以与数据库通信。

当创建一个集合时，它会使用服务工厂和服务设置来创建每个服务的实例。基本集合以及所有扩展它的集合已经具有初始化每个服务的所有详细信息

col = Collection("name", {})
print(f"""Services:
  {col.data=},
  {col.pager=},
  {col.serializer=},
  {col.columns=},
  {col.filters=}""")

assert list(col) == []

此集合没有数据。我们可以初始化一个StaticData实例，并用新的StaticData实例替换集合中现有的数据服务。

每个服务都有一个必需的参数：拥有该服务的集合。所有其他参数都用作服务设置，并且必须按名称传递。请记住，本手册中使用的所有类都在ckanext.collection.utils内部可用。

static_data = StaticData(col, data=[1,2,3])
col.replace_service(static_data)

assert list(col) == [1, 2, 3]

查看Colletion.replace_service。它只接受服务实例。没有必要传递要替换的服务名称 - 集合可以在没有帮助的情况下理解它。并且请注意服务构造函数的第一个参数。它必须是即将使用该服务的集合。某些服务即使传递第一个参数为随机值也可能正常工作，但这是一种例外情况，不应依赖于它。

如果现有的集合不再使用，您将创建一个新的集合，有时您希望从现有的集合中重用服务。只是为了避免创建服务和调用Collection.replace_service，这将为您节省两行代码。在这种情况下，使用集合构造函数的<service>_instance参数。

another_col = Collection("another-name", {}, data_instance=col.data)
assert list(another_col) == [1, 2, 3]

如果您这样做，请确保您不再使用旧集合。您刚刚将其中一个服务从另一个集合中转移出来，因此无法保证带有分离服务的旧集合能够正常工作。

通常最好是自定义服务工厂，而不是传递现有自定义的服务实例。您可以使用集合构造函数的<service>_factory参数来指定用于创建服务实例的类。

col = Collection("name", {}, data_factory=StaticData)
assert list(col) == []

但是，这样我们无法指定data工厂的data属性！别担心，有多种方法可以克服这个问题。首先，服务的所有设置都作为其属性可用。这意味着data设置与服务的data属性相同。如果您可以执行StaticData(..., data=...)，那么您也可以执行service = StaticData(...); service.data = ...

col = Collection("name", {}, data_factory=StaticData)
col.data.data = [1, 2, 3]
assert list(col) == [1, 2, 3]

注意：data服务缓存其数据。如果您已经从StaticData中访问了数据属性，由于缓存，分配新值没有任何效果。您必须在分配后调用col.data.refresh_data()来重建缓存。

但有一种更好的方法。您可以将<service>_settings字典传递给集合构造函数，它将被传递到相应的服务工厂。

col = Collection(
    "name", {},
    data_factory=StaticData,
    data_settings={"data": [1, 2, 3]}
)
assert list(col) == [1, 2, 3]

这对于个别场景工作得很好，但是当您创建大量具有静态数据的集合时，您希望省略一些标准参数。在这种情况下，您应该定义一个新的类，该类扩展了集合并声明了<Service>Factory属性。

class MyCollection(Collection):
    DataFactory = StaticData

col = MyCollection(
    "name", {},
    data_settings={"data": [1, 2, 3]}
)
assert list(col) == [1, 2, 3]

您仍然可以将data_factory传递给MyCollection构造函数以覆盖数据服务工厂。但现在，默认情况下，如果没有明确指定，将使用StaticData。

最后，如果您想创建一个具有特定属性值的服务的子类，即类似于这样

class OneTwoThreeData(StaticData):
    data = [1, 2, 3]

您可以使用Service.with_attributes(attr_name=attr_value)工厂方法。它产生一个新的服务类（工厂），其中指定了属性绑定到静态值。例如，这就是我们定义一个始终包含[1, 2, 3]的集合的方法。

class MyCollection(Collection):
    DataFactory = StaticData.with_attributes(data=[1, 2, 3])

col = MyCollection("name", {})
assert list(col) == [1, 2, 3]

现在创建集合时无需指定data_factory或data_settings。它将始终使用StaticData，其中data设置为[1, 2, 3]。请确保您确实想要这样做，因为现在您无法使用data_settings覆盖数据。

通用逻辑

所有服务都共享一些共同功能。首先，所有服务都包含对使用/拥有服务的集合的引用。只能有一个集合拥有服务。如果您将服务从一个集合移动到另一个集合，您绝不能使用不再拥有服务的旧集合。根据服务的内部实现，它可能无需更改即可正常工作，但我们建议删除此类集合。在任何时候，您都可以通过服务的attached属性获取拥有该服务的集合

col = Collection("name", {})
assert col.data.attached is col
assert col.pager.attached is col
assert col.columns.attached is col

another_col = Collection(
    "another-name", {},
    data_instance=col.data
)
assert col.data.attached is not col
assert col.data.attached is another_col
assert col.data is another_col.data

服务的第二个共同点是设置。让我们使用StaticData进行测试。它有一个可配置的属性（设置）- data。我们可以在创建数据服务实例时直接指定它：StaticData(..., data=DATA)。或者我们可以在创建集合时通过data_settings指定它：StaticCollection("name", {}, data_settings={"data": DATA})。在这两种情况下，DATA都将作为数据服务的data属性可用。但这并不意味着我们可以以这种方式传递任何属性

data = StaticData(col, data=[], not_real=True)
assert hasattr(data, "data")
assert not hasattr(data, "not_real")

为了允许通过设置覆盖属性的值，我们必须将此属性定义为可配置属性。为此，我们需要从ckanext.collection.shared中的configurable_attribute函数

class MyData(StaticData):
    i_am_real = configurable_attribute(False)

data = MyData(col, data=[], i_am_real=True)
assert hasattr(data, "data")
assert hasattr(data, "i_am_real")
assert data.i_am_real is True

configurable_attribute接受属性的默认值（位置参数），或者接受一个名为default_factory的命名函数，该函数每次创建新的服务实例时都会生成默认值。default_factory必须接受一个单一参数 - 在实例化时的新服务

class MyData(StaticData):
    ref = 42
    i_am_real = shared.configurable_attribute(default_factory=lambda self: self.ref * 10)

data = MyData(col, data=[])
assert data.i_am_real == 420

绝不要在default_factory中使用其他可配置属性 - 可配置属性的初始化顺序并非严格定义。在撰写此手册时，可配置属性按字母顺序初始化，但此实现细节可能在将来不通知的情况下更改。

TODO：with_attributes

数据服务

此服务生成集合的数据。每个数据服务都必须

• 是可迭代的，默认情况下迭代所有可用的记录
• 定义total属性，该属性反映可用的记录数量，以便len(list(data)) == data.total
• 定义range(start: Any, end: Any)方法，该方法返回数据的一部分

数据服务的基类Data已经包含了这个逻辑的简单版本。您只需定义一个方法来制作您定制的实现：compute_data。当数据首次被访问时，将调用compute_data。其结果被缓存并用于for循环中的迭代、通过range方法进行切片以及通过total属性进行大小测量。

class CustomData(Data):
    def compute_data(self) -> Any:
        return "abcdefghijklmnopqrstuvwxyz"

col = Collection("name", {}, data_factory=CustomData)
assert list(col) == ["a", "b", "c", "d", "e", "f", "g", "h", "i", "j"]
assert col.data.total == 26
assert col.data.range(-3, None) == "xyz"

如果您需要更复杂的数据源，请确保您定义了__iter__、total和range

class CustomData(Data):
    names = configurable_attribute(default_factory=["Anna", "Henry", "Mary"])

    @property
    def total(self):
        return len(self.names)

    def __iter__(self):
        yield from sorted(self.names)

    def range(self, start: Any, end: Any):
        if not isinstance(start, str) or not isinstance(end, str):
            return []

        for name in self:
            if name < start:
                continue
            if name > end:
                break
            yield name

分页服务

分页服务设置了用于集合的数据的上限和下限。集合使用的默认分页依赖于数值start/end值。但是，您可以定义自定义分页，该分页使用字母顺序或时间顺序的界限，只要您的自定义数据服务的range方法支持这些界限。

标准分页器（ClassicPager）有两个可配置的属性：page（默认

1. 和rows_per_page（默认：10）。

col = StaticCollection("name", {})
assert col.pager.page == 1
assert col.pager.rows_per_page == 10

由于这些值，您在迭代集合时只能看到数据的前10条记录。让我们更改分页设置

col = StaticCollection(
    "name", {},
    data_settings={"data": range(1, 100)},
    pager_settings={"page": 3, "rows_per_page": 6}
)
assert list(col) == [13, 14, 15, 16, 17, 18]

分页细节通常与搜索参数一起传递，并对所需的数据帧有巨大影响。因此，如果缺少pager_settings，则ClassicPager将查找集合参数（集合构造函数的第二个参数）内的设置。但在这种情况下，分页器将只使用具有<collection name>:前缀的项

col = StaticCollection(
    "xxx",
    {"xxx:page": 3, "xxx:rows_per_page": 6},
    data_settings={"data": range(1, 100)}
)
assert list(col) == [13, 14, 15, 16, 17, 18]

col = StaticCollection(
    "xxx",
    {"page": 3, "rows_per_page": 6},
    data_settings={"data": range(1, 100)}
)
assert list(col) == [1, 2, 3, 4, 5, 6, 7, 8, 9, 10]

序列化服务

序列化器将数据转换为文本、二进制或其他任何替代表示形式。例如，如果您想将集合的data服务生成的记录计算到pandas的DataFrame中，您可能需要使用序列化器。

序列化器是列服务的核心用户，因为它包含有关特定数据列的详细信息。序列化器通常会直接迭代数据服务（忽略range方法），以序列化所有可用的记录。

序列化器唯一必需的方法是serialize。此方法必须返回由序列化器转换的数据服务数据。例如，JsonSerializer返回JSON编码的数据字符串。

您不受文本或二进制格式的限制。将数据转换为pandas DataFrame的序列化器是完全有效的序列化器版本。

class NewLineSerializer(Serializer):
    def serialize(self):
        result = ""
        for item in self.attached.data:
            result += str(item) + "\n"

        return result

col = StaticCollection(
    "name", {},
    serializer_factory=NewLineSerializer,
    data_settings={"data": [1, 2, 3]}
)
assert "".join(col.serializer.serialize()) == "1\n2\n3\n"

列服务

此服务包含关于数据记录单独列的附加信息。它定义以下设置

• names: 所有可用的列名。由列服务的其他设置使用
• hidden: 应由序列化器隐藏的列。由序列化器服务使用
• visible: 必须由序列化器显示的列。由序列化器服务使用
• sortable: 支持排序的列。由数据服务使用
• filterable: 支持筛选/分割的列。由数据服务使用
• searchable: 支持部分匹配搜索的列。由数据服务使用
• labels: 列的人类可读标签。由序列化器服务使用

此服务包含由其他服务使用的信息，因此在此处定义附加属性是完全正常的。例如，一些自定义序列化器，将数据序列化为ORC，可以期望在columns服务中存在orc_format属性。因此，您可以将所需的所有附加列相关细节添加到此服务中。

过滤器服务

目前，此服务仅供HTML表格序列化器使用。它有两个可配置的属性static_filters和static_actions。static_filters用于构建数据表的搜索表单。static_actions目前未使用，但您可以将其用于批处理或记录级别操作的详细信息，并使用这些详细信息扩展标准序列化器之一。例如，ckanext-admin-panel定义了内容（删除、恢复、隐藏）的允许操作，并创建引用这些操作的自定义模板。

核心类和用法示例

TBA

数据

TBA

StaticData

TBA

BaseSaData

TBA

StatementSaData

TBA

UnionSaData

TBA

ModelData

TBA

ApiData

TBA

ApiSearchData

TBA

ApiListData

TBA

分页器

TBA

ClassicPager

TBA

列

TBA

过滤器

TBA

序列化器

TBA

CsvSerializer

TBA

JsonlSerializer

TBA

JsonSerializer

TBA

HtmlSerializer

TBA

TableSerializer

TBA

HtmxTableSerializer

TBA

配置设置

# Names of registered collections that are viewable by any visitor, including
# anonymous.
# (optional, default: )
ckanext.collection.auth.anonymous_collections =

# Names of registered collections that are viewable by any authenticated
# user.
# (optional, default: )
ckanext.collection.auth.authenticated_collections =

# Add HTMX asset to pages. Enable this option if you are using CKAN v2.10
# (optional, default: false)
ckanext.collection.include_htmx_asset = false

# Initialize CKAN JS modules every time HTMX fetches HTML from the server.
# (optional, default: false)
ckanext.collection.htmx_init_modules = false

# Import path for serializer used by CSV export endpoint.
# (optional, default: ckanext.collection.utils.serialize:CsvSerializer)
ckanext.collection.export.csv.serializer = ckanext.collection.utils.serialize:CsvSerializer

# Import path for serializer used by JSON export endpoint.
# (optional, default: ckanext.collection.utils.serialize:JsonSerializer)
ckanext.collection.export.json.serializer = ckanext.collection.utils.serialize:JsonSerializer

# Import path for serializer used by JSONl export endpoint.
# (optional, default: ckanext.collection.utils.serialize:JsonlSerializer)
ckanext.collection.export.jsonl.serializer = ckanext.collection.utils.serialize:JsonlSerializer

# Import path for serializer used by `format`-export endpoint.
# (optional, default: )
ckanext.collection.export.<format>.serializer =

集成

ckanext-admin-panel

要启用管理员面板中ckanext-collection的配置表单，启用以下任意模式

scheming.arbitrary_schemas =
    ckanext.collection:ap_config.yaml

许可证

AGPL