私の目標は、Weaviate サポートをpyLodStorage プロジェクトに追加することです 。具体的には、次のサンプル データを使用したいと思います。
https://github.com/WolfgangFahl/pyLoDStorage/blob/master/lodstorage/sample.py
持っている
- 王室の人物のいくつかの記録
- 数千のエントリを持つ都市リスト
- 必要な数のレコードを含むレコードの人為的なリスト
例として。
すべてのデータは表形式です。次のようないくつかの基本的な python 型:
- 力
- ブール
- 整数
- 浮く
- 日にち
- 日付時刻
サポートする必要があります。
プロジェクトhttp://wiki.bitplan.com/index.php/DgraphAndWeaviateTestと、 Docker Compose経由で Weaviate を実行するスクリプトを作成しました。Weaviate Python クライアント 0.4.1 で動作していた Python 単体テストがあります。
https://www.semi.technology/documentation/weaviate/current/how-tos/how-to-create-a-schema.htmlからの情報を使用して、この単体テストをリファクタリングしようとしていますが、方法がわかりませんそれをするために。
CRUD テストを他の 3 つのテストのように実行するために必要なこと: https://github.com/WolfgangFahl/pyLoDStorage/tree/master/tests for
- JSON
- スパークル
- SQL
上記の標準データ型を使用したdictのリスト（別名「テーブル」）の「往復」処理に特に興味があります。だから私は辞書のリストを作成したいと思います:
- いくつかのサンプル レコードを見てスキーマを自動的に導き出す
- スキーマが既に存在するかどうかを確認し、削除する場合は削除します
- スキーマを作成する
- データがすでに存在するかどうかを確認し、削除するかどうかを確認します
- データを追加して保存する
- オプションで、後で参照できるようにスキーマを保存します
- スキーマ情報を使用して、または使用せずにデータを復元する
復元されたデータ (辞書のリスト) が元のデータと同じであることを確認します
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()