PyDatahub

https://pydatahub.readthedocs.io/zh-cn/latest/
Admin
PyDatahub is the official Python SDK for Alibaba Cloud DataHub, providing Python APIs to interact...
Tokens:67,516
Snippets:358
Trust Score:9.6
Update:1 month ago
Show doc for...
Tokens:
Raw
### Source install with git clone and python setup.py install

Source: https://pydatahub.readthedocs.io/zh-cn/latest/installation.html

Installs PyDatahub from source by cloning the repository and running setup.py. For offline environments, dependencies can be installed from the dependency directory using the provided requirements files.

```bash
$ git clone https://github.com/aliyun/aliyun-datahub-sdk-python.git
$ cd aliyun-datahub-sdk-python
$ python setup.py install

```

--------------------------------

### Verify installation with python -c import

Source: https://pydatahub.readthedocs.io/zh-cn/latest/installation.html

Verifies the installation by importing the DataHub module. A successful execution indicates the Python SDK is installed correctly.

```bash
python -c "from datahub import DataHub"

```

--------------------------------

### Quick install with pip install pydatahub

Source: https://pydatahub.readthedocs.io/zh-cn/latest/installation.html

Installs PyDatahub from PyPI. Related dependency packages are installed automatically if they are not already present.

```bash
$ pip install pydatahub

```

--------------------------------

### Offline dependency install with pip install -r

Source: https://pydatahub.readthedocs.io/zh-cn/latest/installation.html

Installs dependencies manually when there is no network access. Run from the dependency directory after cloning the source repository.

```bash
$ cd dependency
$ pip install -r first.txt
$ pip install -r second.txt

```

--------------------------------

### TupleRecord usage example

Source: https://pydatahub.readthedocs.io/zh-cn/latest/api.html

Shows how to create a TupleRecord from a schema and values, then access and modify values using indices or field names. Demonstrates set_value, get_value, len, and has_field operations.

```python
>>> schema = RecordSchema.from_lists(['name', 'id'], [FieldType.STRING, FieldType.STRING])
>>> record = TupleRecord(schema=schema, values=['test', 'test2'])
>>> record.values[0:2]
>>> ['test', 'test2']
>>> record.set_value('name', 'test1')
>>> record.values[0]
>>> 'test1'
>>> record.set_value(0, 'test3')
>>> record.get_value(0)
>>> 'test3'
>>> record.get_value('name')
>>> 'test3'
>>> len(record.values)
2
>>> record.has_field('name')
True
```

--------------------------------

### Create Project with dh.create_project

Source: https://pydatahub.readthedocs.io/zh-cn/latest/tutorial-project.html

Creates a new DataHub project. Requires a project name and comment. The name must be 3-32 characters, start with a letter, contain only letters, digits, and underscores, and is case-insensitive.

```python
dh.create_project(project_name, comment)
```

--------------------------------

### Publish Tuple records with dh.put_records_by_shard

Source: https://pydatahub.readthedocs.io/zh-cn/latest/tutorial-record.html

Complete example of publishing Tuple records to a topic. It waits for shards to be ready, retrieves topic and shard info, builds a RecordSchema, creates three TupleRecord instances (using constructor, values assignment, and set_value), and publishes them with dh.put_records_by_shard. The commented-out section shows the alternative dh.put_records approach. Note the code contains a syntax error in the time_field assignment for record2 and references an undefined shards variable.

```python
from datahub.models import RecordSchema, FieldType, Field

try:
    # block等待所有shard状态ready
    dh.wait_shards_ready(project_name, topic_name)

    topic_result = dh.get_topic(project_name, topic_name)
    print(topic_result)
    print("=======================================\n\n")

    shard_result = dh.list_shard(project_name, topic_name)
    for shard in shard_result.shards:
        print(shard)
    print("=======================================\n\n")

    record_schema = RecordSchema.from_lists(
        ['bigint_field',   'string_field',   'double_field',   'bool_field',      'time_field'       ],
        [FieldType.BIGINT, FieldType.STRING, FieldType.DOUBLE, FieldType.BOOLEAN, FieldType.TIMESTAMP]
    )

    # 建议使用 put_records_by_shard
    records = []

    record0 = TupleRecord(schema=record_schema, values=[1, 'yc1', 10.01, True, 1455869335000000])
    record0.put_attribute('AK', '47')
    records.append(record0)

    record1 = TupleRecord(schema=record_schema)
    record1.values = [1, 'yc1', 10.01, True, 1455869335000000]
    records.append(record1)

    record2 = TupleRecord(schema=record_schema)
    record2.set_value(0, 3)
    record2.set_value(1, 'yc3')
    record2.set_value('double_field', 10.03)
    record2.set_value('bool_field', False)
    record2.set_value('time_field', 1455869335000
    records.append(record2)

    dh.put_records_by_shard(project_name, topic_name, shards[0].shard_id, records)

    # records = []

    # record0 = TupleRecord(schema=record_schema, values=[1, 'yc1', 10.01, True, 1455869335000000])
    # record0.shard_id = shards[0].shard_id
    # record0.put_attribute('AK', '47')
    # records.append(record0)

    # record1 = TupleRecord(schema=record_schema)
    # record1.values = [1, 'yc1', 10.01, True, 1455869335000000]
    # record1.shard_id = shards[1].shard_id
    # records.append(record1)

    # record2 = TupleRecord(schema=record_schema)
    # record2.set_value(0, 3)
    # record2.set_value(1, 'yc3')
    # record2.set_value('double_field', 10.03)
    # record2.set_value('bool_field', False)
    # record2.set_value('time_field', 1455869335000013)
    # record2.shard_id = shards[2].shard_id
    # records.append(record2)

    # put_result = dh.put_records(project_name, topic_name, records)

    print("put tuple %d records" % len(records))
    print("failed records: \n%s" % put_result)
    # failed_indexs如果非空最好对failed record再进行重试
    print("=======================================\n\n")
except DatahubException as e:
    print traceback.format_exc()
    sys.exit(-1)

```

--------------------------------

### list_topic

Source: https://pydatahub.readthedocs.io/zh-cn/latest/api.html

Gets all topics of a project. Returns a list of topics.

```APIDOC
## list_topic

### Description
Get all topics of a project

### Method
list_topic

### Parameters
- **project_name** (string) - Required - project name

### Returns
- **all topics of the project** (datahub.models.ListTopicResult)

### Raises
- `datahub.exceptions.ResourceNotFoundException` if the project not exists
- `datahub.exceptions.InvalidParameterException` if project_name is empty
```

--------------------------------

### Configure DataHub SDK logging to console

Source: https://pydatahub.readthedocs.io/zh-cn/latest/tutorial.html

Configure the DataHub Python SDK to print DEBUG logs to the console. Add this setup before initializing the DataHub client to see debug and error logs from the SDK.

```python
import logging

logger = logging.getLogger('datahub')
sh = logging.StreamHandler()
sh.setLevel(logging.DEBUG)
logger.addHandler(sh)

```

--------------------------------

### Create RecordSchema with from_lists

Source: https://pydatahub.readthedocs.io/zh-cn/latest/api.html

Creates a RecordSchema from parallel lists of field names and FieldType constants. The example shows five fields with types BIGINT, STRING, DOUBLE, BOOLEAN, and TIMESTAMP. The trailing '>>>' indicates an interactive prompt continuation.

```python
>>> schema = RecordSchema.from_lists(
        ['bigint_field'  , 'string_field'  , 'double_field'  , 'bool_field'     , 'time_field'],
        [FieldType.BIGINT, FieldType.STRING, FieldType.DOUBLE, FieldType.BOOLEAN, FieldType.TIMESTAMP]
    )
>>>
```

--------------------------------

### EsConnectorConfig class definition (partial)

Source: https://pydatahub.readthedocs.io/zh-cn/latest/_modules/datahub/models/connector.html

Starts the definition of the EsConnectorConfig class for ElasticSearch, including its docstring listing members. The code is incomplete in the source.

```python
class EsConnectorConfig(ConnectorConfig):
    """
    Connector config for ElasticSearch

    Members:
        index (:class:`str`): index

    """
```

--------------------------------

### Create DataHub instances with different configurations

Source: https://pydatahub.readthedocs.io/zh-cn/latest/api.html

Shows how to instantiate DataHub with default settings, protobuf enabled, and LZ4 compression. Requires valid access_id, access_key, and endpoint; otherwise SignatureNotMatch error is thrown.

```python
>>> datahub = DataHub('**your access id**', '**your access key**', '**endpoint**')
>>> datahub_pb = DataHub('**your access id**', '**your access key**', '**endpoint**', enable_pb=True)
>>> datahub_lz4 = DataHub('**your access id**', '**your access key**', '**endpoint**', compress_format=CompressFormat.LZ4)
>>>
>>> project_result = datahub.get_project('datahub_test')
>>>
>>> print(project_result is None)
>>>
```

--------------------------------

### Initialize DataHub client with access credentials and optional protobuf/LZ4

Source: https://pydatahub.readthedocs.io/zh-cn/latest/tutorial.html

Initialize a DataHub client with your access ID, access key, and endpoint. The first form uses JSON mode for server version <= 2.11, the second uses protobuf for version > 2.11, and the third enables LZ4 compression for put/get record operations.

```python
from datahub import DataHub

access_id = ***your access id***
access_key = ***your access key***
endpoint = ***your datahub server endpoint***
dh = DataHub(access_id, access_key, endpoint, enable_pb=False) # Json mode: for datahub server version <= 2.11
dh = DataHub(access_id, access_key, endpoint) # Use protobuf when put/get record, for datahub server version > 2.11
dh = DataHub(access_id, access_key, endpoint, compress_format=CompressFormat.LZ4) # use lz4 compression when put/get record

```

--------------------------------

### Get Project Details with dh.get_project

Source: https://pydatahub.readthedocs.io/zh-cn/latest/tutorial-project.html

Gets detailed information about a specific DataHub project. Returns a GetProjectResult object with members project_name, comment, create_time, and last_modify_time.

```python
project_result = dh.get_project(project_name)
```

--------------------------------

### datahub.auth.AliyunAccount.get_type()

Source: https://pydatahub.readthedocs.io/zh-cn/latest/api.html

Get account type.

```APIDOC
## [METHOD] datahub.auth.AliyunAccount.get_type()

### Description
Get account type.

### Method
Instance method

### Endpoint
N/A

### Parameters
None

### Response
#### Success Response
- **account type** - The account type.

#### Response Example
```
AccountType.ALIYUN
```

```

--------------------------------

### DataHub.__init__

Source: https://pydatahub.readthedocs.io/zh-cn/latest/_modules/datahub/core.html

Constructor for the DataHub client. Initializes the client with credentials and optional settings.

```APIDOC
## DataHub(access_id, access_key, endpoint=None, enable_pb=True, compress_format=CompressFormat.NONE, **kwargs)

### Description
Main entrance to DataHub. Provides convenient operations on DataHub objects. Requires Aliyun Access ID and Access Key, and endpoint URL.

### Method
Constructor

### Parameters
- **access_id** (str) - Required - Aliyun Access ID
- **access_key** (str) - Required - Aliyun Access Key
- **endpoint** (str) - Optional - Rest service URL
- **enable_pb** (bool) - Optional - Enable protobuf when put/get records, default value is False in version <= 2.11, default value will be True in version >= 2.12
- **compress_format** (CompressFormat) - Optional - Compress format
- **kwargs** - Optional - Additional keyword arguments

### Example
```python
>>> datahub = DataHub('**your access id**', '**your access key**', '**endpoint**')
>>> datahub_pb = DataHub('**your access id**', '**your access key**', '**endpoint**', enable_pb=True)
>>> datahub_lz4 = DataHub('**your access id**', '**your access key**', '**endpoint**', compress_format=CompressFormat.LZ4)
```
```

--------------------------------

### get_subscription

Source: https://pydatahub.readthedocs.io/zh-cn/latest/_modules/datahub/core.html

Gets subscription information.

```APIDOC
## get_subscription

### Description
Get subscription.

### Method
get_subscription

### Parameters
- **project_name** (str) - Required - Project name
- **topic_name** (str) - Required - Topic name
- **sub_id** (str) - Required - Subscription id

### Returns
GetSubscriptionResult - subscription info

### Raises
- ResourceNotFoundException - if the project, topic or subscription not exists
- InvalidParameterException - if project_name, topic_name or sub_id is empty
```

--------------------------------

### 获取metering info：get_metering_info接口

Source: https://pydatahub.readthedocs.io/zh-cn/latest/tutorial-meter.html

使用dh.get_metering_info接口获取指定shard的统计信息，返回GetMeteringInfoResult对象，包含active_time和storage成员。

```python
result = dh.get_metering_info(project_name, topic_name, shard_id)

```

--------------------------------

### datahub.auth.Account.get_type()

Source: https://pydatahub.readthedocs.io/zh-cn/latest/api.html

Get account type, subclass must be provided.

```APIDOC
## [METHOD] datahub.auth.Account.get_type()

### Description
Get account type, subclass must be provided.

### Method
Instance method

### Endpoint
N/A

### Parameters
None

### Response
#### Success Response
- **account type** - The account type.

#### Response Example
```
AccountType.ALIYUN
```

```

--------------------------------

### 初始化offset：init_and_get_subscription_offset

Source: https://pydatahub.readthedocs.io/zh-cn/latest/tutorial-offset.html

初始化offset是开始消费的第一步。最后一个参数可以是一个shard_id，也可以是shard_id的list。返回InitAndGetSubscriptionOffsetResult对象，包含offsets成员（OffsetWithSession对象），其中sequence和timestamp是点位信息，第一次初始化的session_id为0，之后每次初始化都会+1。

```python
init_result = dh.init_and_get_subscription_offset(project_name, topic_name, sub_id, shard_id)
init_result = dh.init_and_get_subscription_offset(project_name, topic_name, sub_id, shard_ids)
```

--------------------------------

### GetRecordsResult.parse_content: parse JSON get records response

Source: https://pydatahub.readthedocs.io/zh-cn/latest/_modules/datahub/models/results.html

Parses a JSON response for the get records API. It reads the StartSeq, Records, NextCursor, and RecordCount fields, creating BlobRecord or TupleRecord instances based on the data type. Requires a 'record_schema' in kwargs when data is not a string.

```python
    @classmethod
    def parse_content(cls, content, **kwargs):
        content = json.loads(to_text(content))

        records = []
        sequence = content['StartSeq']
        for item in content['Records']:
            data = item['Data']
            if isinstance(data, six.string_types):
                record = BlobRecord(values=data)
            else:
                record_schema = kwargs['record_schema']
                record = TupleRecord(schema=record_schema, values=data)
            if 'Attributes' in item:
                record.attributes = item['Attributes']
            record.sequence = sequence
            record.system_time = item['SystemTime']
            sequence += 1
            records.append(record)
        return cls(content['NextCursor'], content['RecordCount'], content['StartSeq'], records)
```

--------------------------------

### init_and_get_subscription_offset

Source: https://pydatahub.readthedocs.io/zh-cn/latest/_modules/datahub/core.html

Opens a subscription offset session and retrieves offset information for the specified shards.

```APIDOC
## init_and_get_subscription_offset

### Description
Opens a subscription offset session and retrieves offset information.

### Method
init_and_get_subscription_offset

### Parameters
- **project_name** (str) - Required - Project name
- **topic_name** (str) - Required - Topic name
- **sub_id** (str) - Required - Subscription id
- **shard_ids** (str) - Required - Shard ids

### Return
InitAndGetSubscriptionOffsetResult - offset info

### Raises
- ResourceNotFoundException - if the project, topic or subscription not exists
- InvalidParameterException - if project_name, topic_name, sub_id or shard_id is empty
```

--------------------------------

### GetPBRecordsResult.parse_content: parse protobuf get records response

Source: https://pydatahub.readthedocs.io/zh-cn/latest/_modules/datahub/models/results.html

Parses a protobuf-framed response for the get records API. It unwraps the frame, verifies the CRC, and raises a DatahubException on mismatch. Builds TupleRecord or BlobRecord instances based on the presence of a record_schema in kwargs, and sets attributes, system_time, and sequence for each record.

```python
    @classmethod
    def parse_content(cls, content, **kwargs):
        crc, compute_crc, pb_str = unwrap_pb_frame(content)
        if crc != compute_crc:
            raise DatahubException('Parse pb response body fail, error: crc check error. crc: %s, compute crc: %s'
                                   % (crc, compute_crc))

        pb_get_record_response = GetRecordsResponse()
        pb_get_record_response.ParseFromString(pb_str)
        next_cursor = pb_get_record_response.next_cursor
        record_count = pb_get_record_response.record_count
        start_sequence = pb_get_record_response.start_sequence
        records = []
        sequence = start_sequence
        for pb_record in pb_get_record_response.records:
            record_schema = kwargs['record_schema']
            if record_schema:
                values = [bp_field_data.value for bp_field_data in pb_record.data.data]
                record = TupleRecord(schema=record_schema)
                record._set_values(values)
            else:
                record = BlobRecord(blob_data=pb_record.data.data[0].value)
            record._attributes = {
                attribute.key: attribute.value for attribute in pb_record.attributes.attributes
            }
            record.system_time = pb_record.system_time
            record.sequence = sequence
            sequence += 1
            records.append(record)
        return cls(next_cursor, record_count, start_sequence, records)
```

--------------------------------

### DataHub constructor

Source: https://pydatahub.readthedocs.io/zh-cn/latest/api.html

Creates the main DataHub client instance for interacting with the DataHub service. Requires access credentials and allows optional configuration for endpoint, protobuf encoding, and compression format.

```APIDOC
## Constructor

### Description
Initializes the main DataHub client with required authentication credentials and optional service settings.

### Method
Constructor

### Signature
`DataHub(access_id , access_key , endpoint=None , enable_pb=True , compress_format= <datahub.models.compress.CompressFormat>)`

### Parameters
- **access_id** (string) - Required - Aliyun Access ID
- **access_key** (string) - Required - Aliyun Access Key
- **endpoint** (string) - Optional - Rest service URL
- **enable_pb** (boolean) - Optional - Enable protobuf for put/get records. Default is False for versions <= 2.11 and True for versions >= 2.12.
- **compress_format** (datahub.models.compress.CompressFormat) - Optional - Compression format, can be NONE, LZ4, ZLIB, or DEFLATE.

### Example
```
>>> datahub = DataHub('**your access id**', '**your access key**', '**endpoint**')
>>> project_result = datahub.get_project('datahub_test')
>>> print(project_result is None)
```
```

--------------------------------

### GetTopicResult class definition

Source: https://pydatahub.readthedocs.io/zh-cn/latest/_modules/datahub/models/results.html

Defines the GetTopicResult class with properties, setters, and parse_content classmethod for results of the get topic API.

```python
[文档]class GetTopicResult(Result):
    """
    Result of get topic api

    Members:
        project_name (:class:`str`): project name

        topic_name (:class:`str`): topic name

        shard_count (:class:`int`) shard count

        life_cycle (:class:`int`) life cycle

        record_type (:class:`datahub.models.RecordType`): record type

        record_schema (:class:`datahub.models.RecordSchema`): record schema

        comment (:class:`str`): project description

        create_time (:class:`int`): create time

        last_modify_time(:class:`int`): last modify time
    """

    __slots__ = ('_project_name', '_topic_name', '_shard_count', '_life_cycle', '_record_type', '_record_schema',
                 '_comment', '_create_time', '_last_modify_time')

    def __init__(self, **kwargs):
        self._project_name = kwargs['project_name'] if 'project_name' in kwargs else ''
        self._topic_name = kwargs['topic_name'] if 'topic_name' in kwargs else ''
        self._shard_count = kwargs['shard_count'] if 'shard_count' in kwargs else 0
        self._life_cycle = kwargs['life_cycle'] if 'life_cycle' in kwargs else 0
        self._record_type = RecordType(kwargs['record_type']) if 'record_type' in kwargs else ''
        self._record_schema = kwargs['record_schema'] if 'record_schema' in kwargs else None
        self._comment = kwargs['comment'] if 'comment' in kwargs else ''
        self._create_time = kwargs['create_time'] if 'create_time' in kwargs else 0
        self._last_modify_time = kwargs['last_modify_time'] if 'last_modify_time' in kwargs else 0

    @property
    def project_name(self):
        return self._project_name

    @project_name.setter
    def project_name(self, value):
        self._project_name = value

    @property
    def topic_name(self):
        return self._topic_name

    @topic_name.setter
    def topic_name(self, value):
        self._topic_name = value

    @property
    def shard_count(self):
        return self._shard_count

    @shard_count.setter
    def shard_count(self, value):
        self._shard_count = value

    @property
    def life_cycle(self):
        return self._life_cycle

    @life_cycle.setter
    def life_cycle(self, value):
        self._life_cycle = value

    @property
    def record_type(self):
        return self._record_type

    @record_type.setter
    def record_type(self, value):
        self._record_type = value

    @property
    def comment(self):
        return self._comment

    @comment.setter
    def comment(self, value):
        self._comment = value

    @property
    def record_schema(self):
        return self._record_schema

    @record_schema.setter
    def record_schema(self, value):
        self._record_schema = value

    @property
    def create_time(self):
        return self._create_time

    @create_time.setter
    def create_time(self, value):
        self._create_time = value

    @property
    def last_modify_time(self):
        return self._last_modify_time

    @last_modify_time.setter
    def last_modify_time(self, value):
        self._last_modify_time = value

    @classmethod
    def parse_content(cls, content, **kwargs):
        content = json.loads(to_text(content))
        record_schema = None
        record_type = RecordType(content['RecordType'])
        if record_type == RecordType.TUPLE:
            record_schema = RecordSchema.from_json_str(content['RecordSchema'])
        return cls(project_name=kwargs['project_name'], topic_name=kwargs['topic_name'],
                   comment=content['Comment'], create_time=content['CreateTime'],
                   last_modify_time=content['LastModifyTime'], life_cycle=content['Lifecycle'],
                   record_type=record_type, record_schema=record_schema, shard_count=content['ShardCount'])
```

--------------------------------

### get_cursor(project_name, topic_name, shard_id, cursor_type, param=-1)

Source: https://pydatahub.readthedocs.io/zh-cn/latest/_modules/datahub/core.html

Gets a cursor for reading records from a shard. Must be called before get_blob_records or get_tuple_records.

```APIDOC
## get_cursor(project_name, topic_name, shard_id, cursor_type, param=-1)

### Description
Get cursor. When you invoke get_blob_records/get_tuple_records, you must be invoke it to get a cursor first.

### Method
get_cursor

### Parameters
- **project_name** (str) - Required - Project name
- **topic_name** (str) - Required - Topic name
- **shard_id** (str) - Required - Shard id
- **cursor_type** (CursorType) - Required - Cursor type, must be of type :class:`datahub.models.CursorType`
- **param** (int) - Optional - Parameter for cursor, default is -1.

### Return
Cursor information (not specified in source)

### Exceptions
Not specified in source
```

--------------------------------

### 查询订阅（get_subscription）

Source: https://pydatahub.readthedocs.io/zh-cn/latest/tutorial-subscription.html

调用get_subscription接口查询订阅详细信息，返回GetSubscriptionResult对象，包含comment、create_time、is_owner、last_modify_time、state、sub_id、topic_name、type等成员，其中state为SubscriptionState枚举对象，分为ACTIVE和INACTIVE。

```python
subscription_result = dh.get_subscription(project_name, topic_name, create_result.sub_id)

```

--------------------------------

### 更新offset：update_subscription_offset

Source: https://pydatahub.readthedocs.io/zh-cn/latest/tutorial-offset.html

更新offset。参数offsets是一个dict，key是shard_id，value可以是OffsetWithSession对象或dict。如果version和session_id发生变化，就会更新失败。当错误信息指出version发生变化，可以通过get_subscription_offset获取最新version继续消费；当错误信息指出session_id发生变化，只能再次使用init_and_get_subscription_offset初始化offset信息。

```python
offsets1 = {
    '0': OffsetWithSession(sequence0, timestamp0, version0, session_id0),
    '1': OffsetWithSession(sequence1, timestamp1, version1, session_id1)
}

offsets2 = {
    '0': {
        'Sequence': 0,
        'Timestamp': 0,
        'Version': 0,
        'SessionId': 0
    },
    '1': {
        'Sequence': 1,
        'Timestamp': 1,
        'Version': 1,
        'SessionId': 1
    }
}

dh.update_subscription_offset(project_name, topic_name, sub_id, offsets1)
dh.update_subscription_offset(project_name, topic_name, sub_id, offsets2)
```

--------------------------------

### datahub.models.subscription 模块完整源代码

Source: https://pydatahub.readthedocs.io/zh-cn/latest/_modules/datahub/models/subscription.html

该代码块是 datahub.models.subscription 模块的完整源代码，包含 SubscriptionState 枚举、Subscription 类和 OffsetBase 类。它定义了订阅的状态、信息字段以及偏移量基础结构。

```python
#!/usr/bin/env python
# -*- coding: utf-8 -*-
# Licensed to the Apache Software Foundation (ASF) under one
# or more contributor license agreements.  See the NOTICE file
# distributed with this work for additional information
# regarding copyright ownership.  The ASF licenses this file
# to you under the Apache License, Version 2.0 (the
# "License"); you may not use this file except in compliance
# with the License.  You may obtain a copy of the License at
#
#     http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing,
# software distributed under the License is distributed on an
# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
# KIND, either express or implied.  See the License for the
# specific language governing permissions and limitations
# under the License.

from __future__ import absolute_import

from enum import Enum

from ..utils import to_str



[文档]class SubscriptionState(Enum):
    """
    Subscription state, there are: ``INACTIVE``, ``ACTIVE``
    """
    INACTIVE = 0
    ACTIVE = 1




[文档]class Subscription(object):
    """
    subscription info

    Members:
        comment (:class:`str`): subscription description

        create_time (:class:`int`): create time

        is_owner (:class:`bool`): owner or not

        last_modify_time (:class:`int`): last modify time

        state (:class:`datahub.models.SubscriptionState`): subscription state

        sub_id (:class:`str`): subscription id

        topic_name (:class:`str`): topic name

        type (:class:`int`): type
    """
    __slots__ = ('_comment', '_create_time', '_is_owner', '_last_modify_time',
                 '_state', '_sub_id', '_topic_name', '_type')

    def __init__(self, comment, create_time, is_owner, last_modify_time,
                 state, sub_id, topic_name, sub_type):
        self._comment = comment
        self._create_time = create_time
        self._is_owner = is_owner
        self._last_modify_time = last_modify_time
        self._state = state
        self._sub_id = sub_id
        self._topic_name = topic_name
        self._type = sub_type

    @property
    def comment(self):
        return self._comment

    @comment.setter
    def comment(self, value):
        self._comment = value

    @property
    def create_time(self):
        return self._create_time

    @create_time.setter
    def create_time(self, value):
        self._create_time = value

    @property
    def is_owner(self):
        return self._is_owner

    @is_owner.setter
    def is_owner(self, value):
        self._is_owner = value

    @property
    def last_modify_time(self):
        return self._last_modify_time

    @last_modify_time.setter
    def last_modify_time(self, value):
        self._last_modify_time = value

    @property
    def state(self):
        return self._state

    @state.setter
    def state(self, value):
        self._state = value

    @property
    def sub_id(self):
        return self._sub_id

    @sub_id.setter
    def sub_id(self, value):
        self._sub_id = value

    @property
    def topic_name(self):
        return self._topic_name

    @topic_name.setter
    def topic_name(self, value):
        self._topic_name = value

    @property
    def type(self):
        return self._type

    @type.setter
    def type(self, value):
        self._type = value

    @classmethod
    def from_dict(cls, dict_):
        return cls(dict_['Comment'], dict_['CreateTime'], dict_['IsOwner'], dict_['LastModifyTime'],
                   SubscriptionState(dict_['State']), dict_['SubId'], dict_['TopicName'], dict_['Type'])

    def to_json(self):
        return {
            'Comment': self._comment,
            'CreateTime': self._create_time,
            'IsOwner': self._is_owner,
            'LastModifyTime': self._last_modify_time,
            'State': self._state.value,
            'SubId': self._sub_id,
            'TopicName': self._topic_name,
            'Type': self._type
        }

    def __repr__(self):
        return to_str(self.to_json())




[文档]class OffsetBase(object):
    """
    offset base class

    Members:
        sequence (:class:`int`): sequence

        timestamp (:class:`int`): timestamp
    """
    __slots__ = ('_sequence', '_timestamp')

    def __init__(self, sequence, timestamp):
        self._sequence = sequence
        self._timestamp = timestamp

    @property
    def sequence(self):
        return self._sequence

    @sequence.setter
    def sequence(self, value):
        self._sequence = value

    @property
    def timestamp(self):
        return self._timestamp

    @timestamp.setter
    def timestamp(self, value):
        self._timestamp = value

    @classmethod
    def from_dict(cls, dict_):
        return cls(dict_['Sequence'], dict_['Timestamp'])
```

--------------------------------

### init_and_get_subscription_offset

Source: https://pydatahub.readthedocs.io/zh-cn/latest/api.html

Opens a subscription offset session. Returns an InitAndGetSubscriptionOffsetResult object.

```APIDOC
## init_and_get_subscription_offset

### Description
Open subscription offset session.

### Method
SDK Method

### Parameters
- **project_name** (string) - Required - Project name
- **topic_name** (string) - Required - Topic name
- **sub_id** (string) - Required - Subscription id
- **shard_ids** (list) - Required - Shard ids

### Returns
- **InitAndGetSubscriptionOffsetResult** - Offset info

### Raises
- `datahub.exceptions.ResourceNotFoundException` - if the project, topic or subscription not exists
- `datahub.exceptions.InvalidParameterException` - if project_name, topic_name, sub_id or shard_id is empty
```

--------------------------------

### Read records with dh.get_blob_records and dh.get_tuple_records

Source: https://pydatahub.readthedocs.io/zh-cn/latest/tutorial-record.html

Reads data from a specified shard starting at a given cursor, with a limit on the number of records. If fewer than limit_num records remain between the cursor and the shard end, the actual count is returned.

```python
dh.get_blob_records(project_name, topic_name, shard_id, cursor, limit_num)
dh.get_tuple_records(project_name, topic_name, shard_id, record_schema, cursor, limit_num)

```

--------------------------------

### OdpsConnectorConfig class with ODPS-specific attributes

Source: https://pydatahub.readthedocs.io/zh-cn/latest/_modules/datahub/models/connector.html

Concrete connector configuration for ODPS, with attributes for project, table, endpoints, credentials, partition mode, time range, and partition config. The constructor defaults partition_config to an empty dict if not provided.

```python
class OdpsConnectorConfig(ConnectorConfig):
    """
    Connector config for odps

    Members:
        project_name (:class:`str`): odps project name

        table_name (:class:`str`): odps table name

        odps_endpoint (:class:`str`): odps endpoint

        tunnel_endpoint (:class:`str`): tunnel endpoint

        access_id (:class:`str`): odps access id

        access_key (:class:`str`): odps access key

        partition_mode (:class:`datahub.models.connector.PartitionMode`): partition mode

        time_range (:class:`int`): time range

        partition_config(:class:`collections.OrderedDict`): partition config
    """

    __slots__ = ('_project_name', '_table_name', '_odps_endpoint', '_tunnel_endpoint', '_access_id',
                 '_access_key', '_partition_mode', '_time_range', '_partition_config')

    def __init__(self, project_name, table_name, odps_endpoint, tunnel_endpoint,
                 access_id, access_key, partition_mode, time_range, partition_config=None):
        if partition_config is None:
            partition_config = {}
        self._project_name = project_name
        self._table_name = table_name
        self._odps_endpoint = odps_endpoint
        self._tunnel_endpoint = tunnel_endpoint
        self._access_id = access_id
        self._access_key = access_key
        self._partition_mode = partition_mode
        self._time_range = time_range
        self._partition_config = partition_config

    @property
    def project_name(self):
        return self._project_name

    @project_name.setter
    def project_name(self, value):
        self._project_name = value

    @property
    def table_name(self):
        return self._table_name

    @table_name.setter
    def table_name(self, value):
        self._table_name = value

    @property
    def odps_endpoint(self):
        return self._odps_endpoint

    @odps_endpoint.setter
    def odps_endpoint(self, value):
        self._odps_endpoint = value

    @property
    def tunnel_endpoint(self):
        return self._tunnel_endpoint
```

--------------------------------

### Fix Python.h missing error with apt-get or yum

Source: https://pydatahub.readthedocs.io/zh-cn/latest/installation.html

Installs the Python development headers needed to resolve the 'Python.h: No such file or directory' error. Provides commands for apt-based and yum-based systems, with separate packages for Python 2.x and 3.x.

```bash
$ sudo apt-get install python-dev   # for python2.x installs
$ sudo apt-get install python3-dev  # for python3.x installs

$ sudo yum install python-devel   # for python2.x installs
$ sudo yum install python34-devel   # for python3.4 installs

```

--------------------------------

### datahub.core module source: DataHub class and project methods

Source: https://pydatahub.readthedocs.io/zh-cn/latest/_modules/datahub/core.html

The complete source of the datahub.core module, defining the DataHub class with methods for project management. The class constructor selects between protobuf and JSON implementations based on the enable_pb flag, and supports optional compression. Methods list_project, create_project, get_project, and update_project delegate to the underlying implementation and raise documented exceptions for invalid parameters or missing resources.

```python
#!/usr/bin/env python
# -*- coding: utf-8 -*-
# Licensed to the Apache Software Foundation (ASF) under one
# or more contributor license agreements.  See the NOTICE file
# distributed with this work for additional information
# regarding copyright ownership.  The ASF licenses this file
# to you under the Apache License, Version 2.0 (the
# "License"); you may not use this file except in compliance
# with the License.  You may obtain a copy of the License at
#
#     http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing,
# software distributed under the License is distributed on an
# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
# KIND, either express or implied.  See the License for the
# specific language governing permissions and limitations
# under the License.

from __future__ import absolute_import

from .utils import type_assert
from .implement import DataHubJson, DataHubPB
from .models import CompressFormat, RecordSchema, FieldType, CursorType, ConnectorType, ConnectorConfig,\
    ConnectorState, ConnectorOffset, SubscriptionState



[文档]class DataHub(object):
    """
    Main entrance to DataHub.

    Convenient operations on DataHub objects are provided.
    Please refer to `DataHub docs <https://datahub.console.aliyun.com/intro/index.html>`_
    to see the details.

    Generally, basic operations such as ``create``, ``list``, ``delete``, ``update`` are provided for each DataHub object.
    Take the ``project`` as an example.

    To create an DataHub instance, access_id and access_key is required, and should ensure correctness,
    or ``SignatureNotMatch`` error will throw.

    :param access_id: Aliyun Access ID
    :param secret_access_key: Aliyun Access Key
    :param endpoint: Rest service URL
    :param enable_pb: enable protobuf when put/get records, default value is False in version <= 2.11, default value will be True in version >= 2.12
    :param compress_format: compress format
    :type compress_format: :class:`datahub.models.compress.CompressFormat`

    :Example:

    >>> datahub = DataHub('**your access id**', '**your access key**', '**endpoint**')
    >>> datahub_pb = DataHub('**your access id**', '**your access key**', '**endpoint**', enable_pb=True)
    >>> datahub_lz4 = DataHub('**your access id**', '**your access key**', '**endpoint**', compress_format=CompressFormat.LZ4)
    >>>
    >>> project_result = datahub.get_project('datahub_test')
    >>>
    >>> print(project_result is None)
    >>>
    """

    def __init__(self, access_id, access_key, endpoint=None, enable_pb=True,
                 compress_format=CompressFormat.NONE, **kwargs):
        if enable_pb:
            self._datahub_impl = DataHubPB(access_id, access_key, endpoint, compress_format, **kwargs)
        else:
            self._datahub_impl = DataHubJson(access_id, access_key, endpoint, compress_format, **kwargs)


[文档]    def list_project(self):
        """
        List all project names

        :return: projects in datahub server
        :rtype: :class:`datahub.models.results.ListProjectResult`
        """
        return self._datahub_impl.list_project()



[文档]    @type_assert(object, str, str)
    def create_project(self, project_name, comment):
        """
        Create a new project by given name and comment

        :param project_name: project name
        :param comment: description of project
        :return: none
        :raise: :class:`datahub.exceptions.InvalidParameterException` if project_name is not valid
        :raise: :class:`datahub.exceptions.ResourceExistException` if project is already existed
        """
        self._datahub_impl.create_project(project_name, comment)



[文档]    @type_assert(object, str)
    def get_project(self, project_name):
        """
        Get a project by given name

        :param project_name: project name
        :return: the right project
        :rtype: :class:`datahub.models.GetProjectResult`
        :raise: :class:`datahub.exceptions.ResourceNotFoundException` if project not exists
        :raise: :class:`datahub.exceptions.InvalidParameterException` if project_name is empty

        .. see also:: :class:`datahub.models.Project`
        """
        return self._datahub_impl.get_project(project_name)



[文档]    @type_assert(object, str, str)
    def update_project(self, project_name, comment):
        """
        Update project comment

        :param project_name: project name
        :param comment: new description of project
        :return: none
        :raise: :class:`datahub.exceptions.ResourceNotFoundException` if project not exists
        :raise: :class:`datahub.exceptions.InvalidParameterException` if project_name is empty or comment is invalid
        """
        return self._datahub_impl.update_project(project_name, comment)

```

--------------------------------

### 获取schema：使用get_topic接口

Source: https://pydatahub.readthedocs.io/zh-cn/latest/tutorial-schema.html

对于已创建的topic，使用get_topic接口获取schema信息，返回的topic_result中包含record_schema属性。

```python
topic_result = dh.get_topic(project_name, topic_name)
record_schema = topic_result.record_schema

```

--------------------------------

### Get cursor with dh.get_cursor using CursorType

Source: https://pydatahub.readthedocs.io/zh-cn/latest/tutorial-record.html

Shows the four ways to obtain a cursor for a shard: OLDEST, LATEST, SEQUENCE, and SYSTEM_TIME. The returned GetCursorResult object's cursor member is used with get_data_record to read data at a specific position.

```python
cursor_result = dh.get_cursor(project_name, topic_name, shard_id, CursorType.OLDEST)
cursor_result = dh.get_cursor(project_name, topic_name, shard_id, CursorType.LATEST)
cursor_result = dh.get_cursor(project_name, topic_name, shard_id, CursorType.SEQUENCE, sequence)
cursor_result = dh.get_cursor(project_name, topic_name, shard_id, CursorType.SYSTEM_TIME, system_time)

```
Tokens:
Raw