Semantica Salesforce 集成指南从 sObject/SOQL 摄取 CRM 数据到知识图谱【免费下载链接】semanticaGraph-Native Infrastructure for Context and Accountable AI Systems项目地址: https://gitcode.com/GitHub_Trending/sema/semantica导读本指南基于 Semantica 的 Salesforce 集成文档 及其底层实现salesforce_ingestor.py系统讲解如何将 Salesforce 中的 Account、Contact、Opportunity 以及自定义对象custom object摄取进 Semantica 的知识图谱流水线。读完本文你将掌握四种认证方式的配置、标准对象/自定义对象/关系字段的摄取方法、自动分页的原始 SOQL 查询、记录转 Semantica 文档格式并直通GraphBuilder的完整链路以及基于源码级实现的参数行为、校验规则与故障排查要点。Semantica 的 Salesforce 集成以simple-salesforce为底层客户端通过统一入口SalesforceIngestor暴露对象摄取、SOQL 查询、schema 发现、文档导出等能力并与 Snowflake、Databricks 等连接器保持一致的 API 设计见 pyproject.toml 中db-salesforce [simple-salesforce1.12.0]的可选依赖声明。安装与依赖Salesforce 支持作为可选依赖引入两种安装方式等价# 方式一通过官方 extra 安装推荐 pip install semantica[db-salesforce] # 方式二单独安装连接器 pip install simple-salesforce1.12.0从源码结构看salesforce_ingestor.py 采用了与其他连接器snowflake、databricks完全一致的可选依赖守卫模式模块始终可以干净导入但若缺少simple-salesforceSALESFORCE_AVAILABLE会置为False直到真正实例化SalesforceConnector时才抛出带有安装指引的ImportError。测试文件 test_salesforce_ingestor.py 也专门覆盖了该场景断言错误信息中同时包含simple-salesforce与db-salesforce字样。基础用法最简单的摄取流程是实例化SalesforceIngestor并调用ingest_sobjectfrom semantica.ingest import SalesforceIngestor import os ingestor SalesforceIngestor( usernameos.getenv(SALESFORCE_USERNAME), passwordos.getenv(SALESFORCE_PASSWORD), security_tokenos.getenv(SALESFORCE_SECURITY_TOKEN), domainos.getenv(SALESFORCE_DOMAIN, login), # test 表示沙盒 ) data ingestor.ingest_sobject(Account, fields[Id, Name, Industry], limit1000) print(fRetrieved {data.row_count} of {data.total_size} matching records) print(fColumns: {data.columns})返回的SalesforceData是一个 dataclass定义于 salesforce_ingestor.py主要字段包括字段含义data清洗后的记录字典列表已剥离simple-salesforce注入的attributes键row_countdata中实际返回的记录数即应用limit之后的数量columns全部记录中出现过的字段名的有序列表sobjectsObject API 名原始 SOQL 跨对象查询时为Nonequery实际执行的 SOQL 查询字符串instance_urlSalesforce 实例基地址如https://myorg.my.salesforce.comtotal_sizeSalesforce REST API 响应中的totalSize即应用 limit 之前匹配查询的记录总数ingested_at对象创建时间戳提示使用环境变量或配合python-dotenv使用.env文件把凭据留在源码之外。SalesforceIngestor()不传任何参数时会自动读取SALESFORCE_*系列环境变量。四种认证方式SalesforceConnector的_validate_auth方法salesforce_ingestor.py在发起任何网络请求之前识别三种完整认证路径缺一不可的组合会立刻抛出ValidationError。下面逐一展开。用户名 / 密码 / 安全令牌标准服务端流程import os from semantica.ingest import SalesforceIngestor ingestor SalesforceIngestor( usernameos.getenv(SALESFORCE_USERNAME), passwordos.getenv(SALESFORCE_PASSWORD), security_tokenos.getenv(SALESFORCE_SECURITY_TOKEN), domainlogin, # 生产环境沙盒请使用 test )运行前设置环境变量export SALESFORCE_USERNAMEyour-usernameexample.com export SALESFORCE_PASSWORDyour-password export SALESFORCE_SECURITY_TOKENyour-security-token这是标准的服务端流程SOAP 登录时安全令牌会被追加到密码之后。令牌可以在Settings → My Personal Information → Reset My Security Token下生成或重置。注意simple-salesforce的构造函数内部会执行一次真实的 SOAP 登录调用因此网络或认证失败会在connect()时抛出而不是在SalesforceConnector.__init__时见connect()的文档注释。JWT Bearer推荐用于 CI/CDimport os from semantica.ingest import SalesforceIngestor ingestor SalesforceIngestor( usernameos.getenv(SALESFORCE_USERNAME), consumer_keyos.getenv(SALESFORCE_CONSUMER_KEY), privatekey_fileos.getenv(SALESFORCE_PRIVATE_KEY_FILE), domainlogin, # 或 test 使用沙盒 )export SALESFORCE_USERNAMEyour-usernameexample.com export SALESFORCE_CONSUMER_KEYyour-connected-app-consumer-key export SALESFORCE_PRIVATE_KEY_FILE/path/to/server.keyJWT bearer 流程使用签名令牌认证全程不传输密码适合服务器到服务器的集成与 CI/CD 流水线。前置条件是在 Salesforce 中配置一个开启了Use digital signatures的 Connected App并在Manage → Profiles / Permission Sets中将预授权用户加入。若希望直接传入 PEM 内容而非文件路径可用SALESFORCE_PRIVATE_KEY替代SALESFORCE_PRIVATE_KEY_FILE。源码中_build_client_kwargs()会优先使用privatekey_file仅在未提供时才回退到privatekey字符串生产环境优先使用文件路径避免把密钥材料写进源码。Session ID Instance URLingestor SalesforceIngestor( session_idos.getenv(SALESFORCE_SESSION_ID), instance_urlos.getenv(SALESFORCE_INSTANCE_URL), )当你的环境已经自行管理 OAuth 令牌生命周期时例如通过 web-server 或 device flow 获取令牌的 Connected App可把访问令牌作为session_id、完整实例 URL如https://myorg.my.salesforce.com作为instance_url传入。这种模式下connect()不做网络调用直接把令牌注入客户端。沙盒Sandboximport os from semantica.ingest import SalesforceIngestor ingestor SalesforceIngestor( usernameos.getenv(SALESFORCE_USERNAME), passwordos.getenv(SALESFORCE_PASSWORD), security_tokenos.getenv(SALESFORCE_SECURITY_TOKEN), domaintest, # 路由到 test.salesforce.com )export SALESFORCE_USERNAMEyour-sandbox-usernameexample.com.sandbox export SALESFORCE_PASSWORDyour-password export SALESFORCE_SECURITY_TOKENyour-security-token export SALESFORCE_DOMAINtest把domainlogin换成domaintest或在环境中设置SALESFORCE_DOMAINtest即可连接 developer 或 full sandbox。此外domain也支持自定义 My Domain 值。环境变量一览所有构造函数参数都有环境变量兜底源码见 salesforce_ingestor.py显式传入的参数优先于环境变量测试test_constructor_arg_overrides_env_var验证了这一点变量参数默认值SALESFORCE_USERNAMEusername—SALESFORCE_PASSWORDpassword—SALESFORCE_SECURITY_TOKENsecurity_token—SALESFORCE_DOMAINdomainloginSALESFORCE_INSTANCE_URLinstance_url—SALESFORCE_SESSION_IDsession_id—SALESFORCE_CONSUMER_KEYconsumer_key—SALESFORCE_PRIVATE_KEY_FILEprivatekey_file—SALESFORCE_PRIVATE_KEYprivatekey—SALESFORCE_API_VERSIONapi_versionsimple-salesforce库默认值文档标注为59.0api_version会被映射为simple-salesforce的version参数转发见测试test_connect_api_version_forwarded_as_version。另外构造函数还接受**config透传给底层库的额外参数如proxies、自定义session。对象摄取Object Ingestion摄取标准对象data ingestor.ingest_sobject( Account, fields[Id, Name, Industry, AnnualRevenue, BillingCity], whereType Customer AND AnnualRevenue 1000000, order_byName ASC, limit5000, ) print(fRetrieved {data.row_count} of {data.total_size} matching records)注意data.row_count是data.data中实际返回的记录数即应用limit之后的数量data.total_size是 Salesforce 的totalSize即 limit 之前匹配查询的记录总数。比较二者即可判断是否取回了全部结果。ingest_sobject会把各片段组装为SELECT ... FROM sobject [WHERE ...] [ORDER BY ...] [LIMIT ...]查询见_build_soqlsalesforce_ingestor.py并遵循nextRecordsUrl自动分页直到取满_query_all方法。limit的边界行为值得注意limitNone不生成LIMIT子句返回所有匹配记录大对象慎用1 ≤ limit ≤ 2000直接把LIMIT嵌入 SOQL由 Salesforce 服务端截断limit 2000SOQL 中不带LIMIT由_query_all在分页循环中通过切片在客户端截断limit0是合法且有定义的请求不要任何记录直接返回空结果避免生成 Salesforce 不接受的LIMIT 0子句并省去一次网络往返。摄取自定义对象自定义对象的 API 名以__c结尾data ingestor.ingest_sobject( My_Custom_Object__c, fields[Id, Name, Custom_Field__c], )关系遍历字段点号记法同样支持data ingestor.ingest_sobject( Contact, fields[Id, Name, Email, Account.Name, Owner.Name], limit10000, )让 Semantica 自动选择字段省略fields时会通过describe()拉取全部可选中字段多一次 API 调用。复合地址与地理定位字段typeaddress、typelocation会被自动排除——原因是 Salesforce 会以INVALID_FIELD拒绝它们在SELECT子句中出现如需这些数据请单独选择其组件字段BillingStreet、BillingCity、Location__Latitude__s等。该排除逻辑实现在_get_all_field_namessalesforce_ingestor.py的_NON_SELECTABLE_TYPES集合中。data ingestor.ingest_sobject(Opportunity)原始 SOQL 摄取ingest_query可以把任意合法 SOQL 原样传给 Salesforce REST API分页由连接器自动处理data ingestor.ingest_query( SELECT Id, Name, StageName, Amount, CloseDate, Account.Name, Owner.Name FROM Opportunity WHERE IsClosed false ORDER BY CloseDate ASC ) print(fOpen opportunities: {data.row_count})查询字符串会原封不动地传给 SalesforceSOQL 的正确性与安全性由调用方负责。⚠️ 警告ingest_query不会校验或净化 SOQL 字符串。当查询由应用可控输入拼装时请改用ingest_sobject——它会分别校验 sObject 名、字段名以及 WHERE / ORDER BY 片段。ingest_sobject的安全校验值得一提模块顶部定义了一系列正则与黑名单salesforce_ingestor.py。sObject 名必须匹配^[A-Za-z][A-Za-z0-9_]*(__c|__mdt|__e|__b|__x|__ka|__kav|__r)?$字段名逐组件校验Owner.Name的每个点分隔组件单独匹配WHERE 片段先对单引号字符串字面量做掩码SOQL 用转义单引号而非反斜杠再对语句分隔符、注释标记和union/insert/update/delete/drop/alter/create/exec/grant/revoke等注入原语做黑名单匹配。全部校验在连接建立之前完成坏输入只抛ValidationError、不产生任何网络调用。文档导出Document Export把摄取结果转换为 Semantica 文档格式即可直接喂给GraphBuilderdocuments ingestor.export_as_documents( data, id_fieldId, # 默认Salesforce 18 位记录 Id text_fields[Name, Description], # 省略则以空格连接所有字符串字段 ) print(fCreated {len(documents)} documents) # 每个文档的结构 # { # id: 001xx000003GYk2AAG, # text: Acme Corp Enterprise software company, # metadata: { # source: salesforce, # sobject: Account, # instance_url: https://myorg.my.salesforce.com, # row_data: { ... 完整清洗后的记录 ... } # } # }export_as_documentssalesforce_ingestor.py与SnowflakeIngestor、DatabricksIngestor输出完全一致的{id, text, metadata}结构。text_fields缺省时只连接字符串类型且非空的值id_field缺省为IdSalesforce 规范 18 位记录 ID字段缺失时退化为记录索引。随后直通知识图谱构建from semantica.kg import GraphBuilder builder GraphBuilder() kg builder.build(documents)完整的 KG 能力可参考 知识图谱参考文档。对象与 Schema 发现# 列出当前用户可访问的全部 sObject sobject_names ingestor.list_sobjects() print(sobject_names[:10]) # [Account, Case, Contact, ...] # 查看某个 sObject 的字段信息 schema ingestor.get_sobject_schema(Account) for field in schema[fields]: print(f{field[name]}: {field[type]} (nillable{field[nillable]}))list_sobjects底层调用sf.describe()GET /services/data/vXX.0/sobjects并返回排序后的 API 名列表get_sobject_schema调用sf.SObjectName.describe()返回的 schema 字典结构与SnowflakeIngestor.get_table_schema()对齐包含name、label、fields每项含name/type/label/nillable/length与queryable键。上下文管理器Context Manager长时间运行的批处理任务推荐使用上下文管理器——进入时建立一条连接退出时关闭with块内的每次摄取调用都复用同一个已认证会话with SalesforceIngestor( usernameos.getenv(SALESFORCE_USERNAME), passwordos.getenv(SALESFORCE_PASSWORD), security_tokenos.getenv(SALESFORCE_SECURITY_TOKEN), ) as sf: accounts sf.ingest_sobject(Account, limit10000) contacts sf.ingest_sobject(Contact, limit10000) sobjects sf.list_sobjects()这是由SalesforceIngestor.__enter__/__exit__salesforce_ingestor.py与SalesforceConnector.connect()的连接复用逻辑共同保证的connect()在客户端已打开时直接返回现有实例避免重复 SOAP/OAuth 往返与资源泄漏测试test_context_manager_connects_only_once断言底层构造器只被调用一次即使块内抛出异常__exit__也会通过close()释放连接。disconnect()会关闭simple-salesforce内部的requests.Session并将客户端引用置空且可安全重复调用。便捷函数 ingest_salesforceingest_salesforce()定义于 methods.py并通过 ingest/init.py 导出把连接 摄取 返回压缩为一次调用from semantica.ingest import ingest_salesforce # 摄取记录凭据来自环境变量 data ingest_salesforce( methodsobject, sobject_nameAccount, fields[Id, Name, Industry], limit500, ) # 执行原始 SOQL data ingest_salesforce( methodquery, soqlSELECT Id, Name FROM Contact WHERE IsActive true, ) # 一步完成摄取 导出为文档 docs ingest_salesforce( methoddocuments, sobject_nameAccount, text_fields[Name, Description], limit1000, ) # 列出可访问的 sObject sobject_names ingest_salesforce(methodlist_sobjects)支持的方法包括sobject默认需要sobject_name、query需要soql、list_sobjects、schema需要sobject_name返回字段元数据字典与documents摄取后立即转文档。凭据也可以通过source字典传入其键与连接器构造函数一一对应配置优先级从低到高为全局方法配置 →source字典凭据 → 关键字参数凭据见 methods.py。或者使用统一的ingest()分发器from semantica.ingest import ingest result ingest( None, source_typesalesforce, methodsobject, sobject_nameAccount, fields[Id, Name], limit500, ) data result[data] # SalesforceData连接测试与故障排查先用SalesforceConnector.test_connection()快速验证凭据与网络import os from semantica.ingest import SalesforceConnector connector SalesforceConnector( usernameos.getenv(SALESFORCE_USERNAME), passwordos.getenv(SALESFORCE_PASSWORD), security_tokenos.getenv(SALESFORCE_SECURITY_TOKEN), ) if not connector.test_connection(): print(Connection failed: check username, password, security token, and domain)test_connection()会认证并调用轻量的/limits/端点只读、不触碰 CRM 数据来证明会话有效临时连接用完即关且不会误关调用前已存在的连接相关行为均有测试覆盖。认证失败的常见原因错误的 domain生产组织用domainlogin沙盒用domaintest。安全令牌过期在Settings → Reset My Security Token下重置新令牌会发送到你的邮箱。IP 限制组织的可信 IP 范围可能拦截来源 IP检查Setup → Network Access。API 访问未启用确保关联的 Profile 具有API Enabled权限。值得一提的安全设计凭据密码、令牌、session ID只存放在私有属性中从不进入日志或异常消息认证失败抛出的ProcessingError会抑制原始异常其可能包含用户名等敏感信息。测试文件 test_salesforce_ingestor.py 中有一整组TestSalesforceConnectorSecrets用例断言密码、令牌、session ID 均不会出现在repr()或异常文本里。连接失败后_client也保持为None不会残留半初始化状态。数据清洗与分页原理SalesforceData中的data并非 REST API 的原始返回而是经过_convert_rows/_clean_recordsalesforce_ingestor.py递归处理后的结果剥离attributes键simple-salesforce会向每条记录及嵌套的关系子对象注入attributes元数据字典清洗时全部丢弃递归扁平化嵌套关系对象例如{Owner: {attributes: {...}, Name: Alice}}变为{Owner: {Name: Alice}}类型规范化datetime对象转为 ISO-8601 字符串其余不可序列化类型经由str()转换保证结果可直接 JSON 序列化并进入后续流水线。分页方面_query_all以result[done]为终止条件循环调用client.query_more(next_url, identifier_is_urlTrue)拉取后续页并追加记录直到取满limit或结果耗尽total_size取第一页响应的totalSize因此能精确回答我还有多少记录没取到。后续延伸Ingest 模块参考完整的SalesforceIngestorAPI 与全部摄取器。Snowflake 集成设计相近的关系型数仓连接器。Databricks 集成Lakehouse 连接器。安装指南全部可选依赖 extra。知识图谱参考基于摄取到的 Salesforce 数据构建知识图谱。【免费下载链接】semanticaGraph-Native Infrastructure for Context and Accountable AI Systems项目地址: https://gitcode.com/GitHub_Trending/sema/semantica创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考