Tordor LiQiuYu
0 Followers · 0 Following
  • Joined on 2025-07-25
Repositories
2
 Projects Packages Public Activity Starred Repositories

ylhp-boss-hr (0.1.0)

Published 2026-03-05 11:56:30 +08:00 by LiQiuYu

Installation

pip install --index-url  --extra-index-url https://pypi.org/simple ylhp-boss-hr

About this package

BOSSHRSDK

XSY Neocrm API Client

这是一个用于与 Neocrm API 交互的 Python 客户端库。它简化了 Neocrm REST API 和批量操作的调用，提供了方便的认证管理和数据交互功能。

功能特性

  • 自动化认证：自动处理 OAuth2 认证流程，包括获取和刷新访问令牌，确保 API 请求的有效性。
  • 灵活的 REST API 封装：支持对 Neocrm XObjects 进行 GET, POST, PUT (PATCH), DELETE 等标准 RESTful 操作。
  • 全面的批量操作支持：提供了一套完整的批量作业和批量任务管理功能，包括：
    • 创建和管理批量作业。
    • 创建和查询批量任务。
    • 获取批量任务的详细信息和执行结果。
  • 模块化设计：通过继承 XSYBase 和 XSYBulk 类，实现了清晰的职责分离，便于扩展和维护。

安装

（请在此处添加您的安装说明，例如 pip install 或 poetry add）

# 示例：如果您使用 Poetry
poetry install

使用方法

认证

您需要提供您的 Neocrm 客户端 ID、客户端密钥、用户名和密码来初始化 Neocrm 客户端：

import asyncio
from neocrmsdk import Neocrm

async def authenticate_example():
    data = {
        "grant_type": "password",
        "client_id": "YOUR_CLIENT_ID",
        "client_secret": "YOUR_CLIENT_SECRET",
        "username": "YOUR_USERNAME",
        "password": "YOUR_PASSWORD",
    }
    client = Neocrm(**data)
    # 访问 access_token 会触发自动认证和刷新
    token = await client.access_token
    print(f"Access Token: {token}")

if __name__ == "__main__":
    asyncio.run(authenticate_example())

XObjects 操作

可以通过 _get_xobjects、_get_object_description、_create_object、_update_object、_delete_object 和 _get_object 方法来操作 Neocrm 的 XObjects。

import asyncio
from src.Neocrm import Neocrm

async def xobject_example():
    # ... 客户端初始化代码 ...
    client = Neocrm(...
    
    # 获取所有 XObjects
    xobjects = await client._get_xobjects()
    print("XObjects:", xobjects)
    
    # 获取特定 XObject 的描述
    # lead_description = await client._get_object_description("lead") # 示例，如果 field_lead 属性存在
    # print("Lead Description:", lead_description)
    
    # 创建一个新的 Lead 对象
    # new_lead_data = {"name": "Test Lead", "status": "New"}
    # created_lead = await client._create_object("lead", new_lead_data)
    # print("Created Lead:", created_lead)

if __name__ == "__main__":
    asyncio.run(xobject_example())

批量作业和任务

该库提供了对 Neocrm 批量 API 的全面支持。

import asyncio
import json
from neocrmsdk import Neocrm
from neocrmsdk.param.bulk import CreateBulkBatchRequest, CreateBulkJobRequest

async def bulk_example():
    # ... 客户端初始化代码 ...
    client = Neocrm(...

    # 创建批量作业
    create_job_request = CreateBulkJobRequest(
        operation="query",
        object="lead"
    )
    job_response = await client._create_bulk_job(create_job_request)
    job_id = job_response.result.id
    print(f"创建批量作业成功，Job ID: {job_id}")

    # 创建异步任务
    create_batch_request = CreateBulkBatchRequest(
        jobId=job_id,
        datas=[
            {"q": "select id from lead limit 1"}
        ]
    )
    batch_response = await client._create_bulk_batch(create_batch_request)
    batch_id = batch_response.result.batchId
    print(f"创建异步任务成功，Batch ID: {batch_id}")

    # 获取批量任务结果
    batch_result = await client._get_bulk_batch_result(batch_id=batch_id)
    print("批量任务结果:", json.dumps(batch_result.result.records, indent=2, ensure_ascii=False))

    # 获取所有批量作业
    all_jobs = await client._get_all_bulk_job()
    print("所有批量作业:", json.dumps(all_jobs.result.model_dump(), indent=2, ensure_ascii=False))

if __name__ == "__main__":
    asyncio.run(bulk_example())

模块结构

  • : 入口文件，包含使用示例。
  • : 继承 XSYBase 和 XSYBulk，是客户端的入口类。
  • : 基础类，处理认证、HTTP 请求和通用的 XObjects 操作。
  • : 处理 Neocrm 批量 API 的相关操作。
  • : 包含 API 请求和响应的数据模型。

Requirements

Requires Python: >=3.12
Details
PyPI
2026-03-05 11:56:30 +08:00
4
16 KiB
Assets (2)
ylhp_boss_hr-0.1.0-py3-none-any.whl 7.9 KiB
ylhp_boss_hr-0.1.0.tar.gz 8.2 KiB
Versions (40) View all
1.51 2026-03-31
1.50 2026-03-31
1.49 2026-03-31
1.48 2026-03-30
1.48.dev0+g674c430ae.d20260330 2026-03-30