Spaces:
Runtime error
Runtime error
File size: 2,099 Bytes
a4b89be |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 |
from azure.cosmos import CosmosClient, PartitionKey
from config import ENDPOINT, CREDENTIAL, DATABASE, CONTAINER_COSMOS
from datetime import date, datetime
import json
_client = CosmosClient(
url=ENDPOINT,
credential=CREDENTIAL,
)
database = _client.create_database_if_not_exists(DATABASE)
_container = database.create_container_if_not_exists(
CONTAINER_COSMOS,
partition_key=PartitionKey("/user_id")
)
def json_serial(obj):
"""JSON serializer for objects not serializable by default json code"""
if isinstance(obj, (datetime, date)):
return obj.isoformat()
raise TypeError ("Type %s not serializable" % type(obj))
def upsert_item(user_id, file_name, history, chatbot):
response = _container.upsert_item(
body={
"id": file_name,
"user_id": user_id,
"date": json.dumps(datetime.utcnow(), default=json_serial),
"history": history,
"chatbot": chatbot
}
)
message = f'Upsert {file_name} succesfully'
return message
def read_item(user_id, file_name):
response = _container.read_item(item=file_name, partition_key=user_id)
return response
def query_items(user_id, file_name):
response = list(_container.query_items(
query="SELECT * FROM r WHERE r.user_id=@user_id AND r.id=@id",
parameters=[
{"name": "@user_id", "value": user_id}, {"name": "@id", "value": file_name}
],
enable_cross_partition_query=True
))
return response
def query_item(user_id):
response = list(_container.query_items(
query="SELECT * FROM r WHERE r.user_id=@user_id",
parameters=[
{"name": "@user_id", "value": user_id}
],
enable_cross_partition_query=True
))
return response
def delete_items(user_id, file_name):
response = _container.delete_item(item=file_name, partition_key=user_id)
message = f'Delete {file_name} succesfully'
return message
if __name__ == '__main__':
mes = query_item("khanh")
docs = [m["id"] for m in mes]
print(docs)
|