我的目标是为pyLodStorage 项目添加 Weaviate 支持 具体来说，我想使用以下示例数据：
https://github.com/WolfgangFahl/pyLoDStorage/blob/master/lodstorage/sample.py
其中有
- 一些皇室成员的记录
- 包含数千个条目的城市列表
- 一个人为的记录列表，其中包含您希望的尽可能多的记录
作为例子。
所有数据都是表格。一些基本的python类型，如：
- 字符串
- 布尔
- 整数
- 漂浮
- 日期
- 约会时间
需要支持。
我创建了项目http://wiki.bitplan.com/index.php/DgraphAndWeaviateTest和一个通过 docker compose运行 Weaviate 的脚本。有一个 python 单元测试用于与 Weaviate Python 客户端 0.4.1 一起工作
我正在尝试使用来自https://www.semi.technology/documentation/weaviate/current/how-tos/how-to-create-a-schema.html的信息来重构这个单元测试，但不知道如何去做吧。
需要做些什么才能让 CRUD 测试像在其他三个测试中一样运行： https : //github.com/WolfgangFahl/pyLoDStorage/tree/master/tests
- JSON
- SPARQL
- SQL
我对具有上述标准数据类型的字典列表（又名“表”）的“往返”处理特别感兴趣。所以我想创建一个字典列表，然后：
- 通过查看一些示例记录自动派生模式
- 检查架构是否已经存在，如果删除它
- 创建架构
- 检查数据是否已经存在，如果删除它
- 添加数据并存储
- 可选地存储模式以供进一步参考
- 使用或不使用架构信息恢复数据
检查恢复的数据（字典列表）是否与原始数据相同
Created on 2020-07-24
@author: wf
'''
import unittest
import weaviate
import time
#import getpass
class TestWeaviate(unittest.TestCase):
# https://www.semi.technology/documentation/weaviate/current/client-libs/python.html
def setUp(self):
self.port=8153
self.host="localhost"
#if getpass.getuser()=="wf":
# self.host="zeus"
# self.port=8080
pass
def getClient(self):
self.client=weaviate.Client("http://%s:%d" % (self.host,self.port))
return self.client
def tearDown(self):
pass
def testRunning(self):
'''
make sure weaviate is running
'''
w=self.getClient()
self.assertTrue(w.is_live())
self.assertTrue(w.is_ready())
def testWeaviateSchema(self):
''' see https://www.semi.technology/documentation/weaviate/current/client-libs/python.html '''
w = self.getClient()
#contains_schema = w.schema.contains()
try:
w.create_schema("https://raw.githubusercontent.com/semi-technologies/weaviate-python-client/master/documentation/getting_started/people_schema.json")
except:
pass
entries=[
[ {"name": "John von Neumann"}, "Person", "b36268d4-a6b5-5274-985f-45f13ce0c642"],
[ {"name": "Alan Turing"}, "Person", "1c9cd584-88fe-5010-83d0-017cb3fcb446"],
[ {"name": "Legends"}, "Group", "2db436b5-0557-5016-9c5f-531412adf9c6" ]
]
for entry in entries:
dict,type,uid=entry
try:
w.create(dict,type,uid)
except weaviate.exceptions.ThingAlreadyExistsException as taee:
print ("%s already created" % dict['name'])
pass
def testPersons(self):
return
w = self.getClient()
schema = {
"actions": {"classes": [],"type": "action"},
"things": {"classes": [{
"class": "Person",
"description": "A person such as humans or personality known through culture",
"properties": [
{
"cardinality": "atMostOne",
"dataType": ["text"],
"description": "The name of this person",
"name": "name"
}
]}],
"type": "thing"
}
}
w.create_schema(schema)
w.create_thing({"name": "Andrew S. Tanenbaum"}, "Person")
w.create_thing({"name": "Alan Turing"}, "Person")
w.create_thing({"name": "John von Neumann"}, "Person")
w.create_thing({"name": "Tim Berners-Lee"}, "Person")
def testEventSchema(self):
'''
https://stackoverflow.com/a/63077495/1497139
'''
return
schema = {
"things": {
"type": "thing",
"classes": [
{
"class": "Event",
"description": "event",
"properties": [
{
"name": "acronym",
"description": "acronym",
"dataType": [
"text"
]
},
{
"name": "inCity",
"description": "city reference",
"dataType": [
"City"
],
"cardinality": "many"
}
]
},
{
"class": "City",
"description": "city",
"properties": [
{
"name": "name",
"description": "name",
"dataType": [
"text"
]
},
{
"name": "hasEvent",
"description": "event references",
"dataType": [
"Event"
],
"cardinality": "many"
}
]
}
]
}
}
client = self.getClient()
if not client.contains_schema():
client.create_schema(schema)
event = {"acronym": "example"}
client.create(event, "Event", "2a8d56b7-2dd5-4e68-aa40-53c9196aecde")
city = {"name": "Amsterdam"}
client.create(city, "City", "c60505f9-8271-4eec-b998-81d016648d85")
time.sleep(2.0)
client.add_reference("c60505f9-8271-4eec-b998-81d016648d85", "hasEvent", "2a8d56b7-2dd5-4e68-aa40-53c9196aecde")
if __name__ == "__main__":
#import sys;sys.argv = ['', 'Test.testName']
unittest.main()