docker_webserver #10
@@ -0,0 +1,27 @@
|
||||
from pathlib import Path
|
||||
from dotenv import load_dotenv
|
||||
from pymongo import MongoClient
|
||||
from typing import List
|
||||
import os
|
||||
|
||||
from src.database import VisualCommunication, connect
|
||||
|
||||
if __name__ == "__main__":
|
||||
# prepare env vars
|
||||
env_path = Path(__file__).parent.parent / "local.env"
|
||||
assert env_path.exists()
|
||||
load_dotenv(env_path)
|
||||
os.environ["MONGO_HOST"] = "localhost"
|
||||
# connect to database
|
||||
collection, db, client = connect()
|
||||
print(client.server_info())
|
||||
# download images
|
||||
data = None
|
||||
for data in collection.find().limit(3):
|
||||
if data is None:
|
||||
print("no document found")
|
||||
break
|
||||
vis_com: VisualCommunication = VisualCommunication.model_validate(data)
|
||||
print(repr(vis_com))
|
||||
if data is not None:
|
||||
print(vis_com.image)
|
||||
@@ -1,6 +1,6 @@
|
||||
from pathlib import Path
|
||||
from dotenv import load_dotenv
|
||||
from pymongo import MongoClient
|
||||
from pymongo.errors import DuplicateKeyError
|
||||
import os
|
||||
|
||||
from src.database import VisualCommunication, connect
|
||||
@@ -15,12 +15,19 @@ if __name__ == "__main__":
|
||||
vis_com_list = [VisualCommunication.from_file(path) for path in img_path_list]
|
||||
for vis_com in vis_com_list:
|
||||
print(repr(vis_com))
|
||||
# upload images to database
|
||||
env_path = test_dir.parent / "mongodb.env"
|
||||
# prepare env vars
|
||||
env_path = test_dir.parent / "local.env"
|
||||
assert env_path.exists()
|
||||
load_dotenv(env_path)
|
||||
os.environ["MONGO_HOST"] = "localhost"
|
||||
# connect to database
|
||||
collection, db, client = connect()
|
||||
print(client.server_info())
|
||||
# upload images
|
||||
for vis_com in vis_com_list:
|
||||
result = collection.insert_one(vis_com.model_dump_json())
|
||||
print(f"inserted document: {result}")
|
||||
try:
|
||||
result = collection.insert_one(vis_com.model_dump())
|
||||
except DuplicateKeyError as exc:
|
||||
print("ignoring:\n", exc)
|
||||
else:
|
||||
print(f"inserted document: {result}")
|
||||
|
||||
Reference in New Issue
Block a user