本体函数Python functions [Beta]类型参考

注意:以下翻译的准确性尚未经过验证。这是使用 AIP ↗ 从原始英文文本进行的机器翻译。

类型参考

Beta

Python 函数目前处于测试阶段,可能无法在所有注册中使用。

以下是当前支持的函数 API 类型的完整列表,它们对应的 Python 类型,以及是否可以使用其对应的 Python 类型而不是函数 API 类型进行声明:

函数 API 类型可以声明为 Python 类型吗?对应的 Python 类型
Arraylist
Binarybytes
Booleanbool
Byteint*
Datedatetime.date
Decimaldecimal.Decimal
Doublefloat*
Floatfloat
Integerint
Longint*
Mapdict
Setset
Shortint*
字符串str
Timestampdatetime.datetime
结构体/自定义类型dataclass.dataclass (示例)

虽然 IntegerLong 都对应于 Python 类型 int,但在函数签名中直接标记为 int 的字段将注册为 Integer 类型。因此,我们建议使用 API 中的 IntegerLong 类型来注册数值数据类型。类似的指导适用于 FloatDouble;如果 Python 类型 float 直接在您的函数签名中,则默认将注册为 Float

下面显示了另一个带有输入的函数示例:

Copied!
1 2 3 4 5 6 7 from functions.api import function, Long, String, Timestamp @function def get_end_day_of_week(start_time: Timestamp, elapsed_millis: Long) -> String: # 这里是函数逻辑。这个函数的目的是根据给定的开始时间(Timestamp)和经过的毫秒数(Long), # 计算并返回一周的结束日期(String)。 pass

如上面的类型表所示,此函数也可以仅使用内置Python类型声明:

Copied!
1 2 3 4 5 6 7 from functions.api import function from datetime import datetime @function def get_end_day_of_week(start_time: datetime, elapsed_millis: int) -> str: # 这里是函数逻辑 pass

在这个代码片段中,我们定义了一个装饰函数 @function,并创建了一个函数 get_end_day_of_week,它接受两个参数:start_time (类型为 datetime)和 elapsed_millis(类型为整数,表示经过的毫秒数)。该函数应该返回一个字符串类型的结果,表示一周中的结束日期。函数的具体实现逻辑尚未编写。 此外,您可以结合使用内置类型和API类型:

Copied!
1 2 3 4 5 6 7 from functions.api import function, Long, String from datetime import datetime @function def get_end_day_of_week(start_time: datetime, elapsed_millis: Long) -> String: # 在这里编写函数逻辑 pass

自定义类型和结构体

由其他支持类型组成的自定义Python类也可以被用于在函数签名中。API包没有明确的Custom类型;相反,自定义类型被声明为用户定义的Python类。

要成为有效的自定义类型,类必须符合以下要求:

  • 类的所有字段必须有类型注解。
  • 字段类型必须是支持的类型;可以使用原始的API类型或原生Python类型(如上表所定义)。
  • __init__方法必须仅接受与字段具有相同名称和类型注解的命名参数。

dataclasses.dataclass装饰器可用于自动生成符合这些要求的__init__方法。

使用dataclasses.dataclass的自定义类型示例

以下是使用dataclasses.dataclass的示例:

Copied!
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 from dataclasses import dataclass from functions.api import Float, Integer, String, function @dataclass class InventoryItem: """Class for keeping track of an item in inventory. 用于记录库存中商品的类。 """ name: String # 商品名称 unit_price: Float # 单价 quantity_on_hand: Integer = 0 # 库存数量,默认为0 def total_cost(self) -> Float: # 计算库存商品的总成本 return self.unit_price * self.quantity_on_hand @function def custom_type_with_init_from_decorator(inventory_item: InventoryItem) -> Float: # 使用装饰器定义的函数,接收一个InventoryItem对象并返回其总成本 return inventory_item.total_cost()

使用API类型的自定义类型示例(不使用dataclass.dataclass

以下是不使用dataclass.dataclass而使用API类型的示例:

Copied!
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 from functions.api import function, String, Timestamp class Event: timestamp: Timestamp # 时间戳属性,类型为Timestamp message: String # 消息属性,类型为String def __init__( self, timestamp: Timestamp, message: String ): self.timestamp = timestamp # 初始化时间戳 self.message = message # 初始化消息 @function def get_event_message(event: Event) -> String: return event.message # 返回事件的消息属性

此代码定义了一个Event类,用于表示带有时间戳和消息的事件。get_event_message函数用于从Event对象中提取和返回消息。

下一步

了解如何设置Python函数库,或了解更多关于在Pipeline搭建器Workshop中使用Python函数的信息。