Skip to content

Instantly share code, notes, and snippets.

@skeptrunedev
Last active November 30, 2023 07:31
Show Gist options
  • Save skeptrunedev/2a5493b105b527e37d5f4babc81d380b to your computer and use it in GitHub Desktop.
Save skeptrunedev/2a5493b105b527e37d5f4babc81d380b to your computer and use it in GitHub Desktop.
Python script to migrate one Arguflow instance to another using postgres of origin instance only
import psycopg2
from dotenv import load_dotenv
import os
import json
import requests
# Load the .env file
load_dotenv()
api_key = os.environ.get("API_KEY")
api_url = os.environ.get("API_URL")
origin_db_url = os.environ.get("ORIGIN_DB_URL")
class Card:
def __init__(self, card_html, tag_set, tracking_id, metadata_dict):
self.card_html = card_html
self.tag_set = tag_set
self.tracking_id = tracking_id
self.metadata = metadata_dict
if not self.metadata:
print("Missing metadata.")
exit(1)
def to_json(self):
def replace_nan_none(obj):
if isinstance(obj, float) and (obj != obj or obj is None):
return ""
if obj is None:
return ""
if isinstance(obj, dict):
return {key: replace_nan_none(value) for key, value in obj.items()}
if isinstance(obj, list):
return [replace_nan_none(item) for item in obj]
return obj
json_dict = {
key: replace_nan_none(value) for key, value in self.__dict__.items()
}
return json.dumps(json_dict, sort_keys=True, default=str)
def send_post_request(self):
url = f"{api_url}/card"
payload = self.to_json()
headers = {"Content-Type": "application/json", "Authorization": api_key}
req_result = requests.post(url, data=payload, headers=headers)
if req_result.status_code != 200:
req_error = req_result.text
print(req_error)
conn = psycopg2.connect(origin_db_url)
cur = conn.cursor()
cur.execute("SELECT DISTINCT ON (metadata) * FROM card_metadata")
while True:
# Fetch 20 rows
rows = cur.fetchmany(20)
# If no more rows are available, break the loop
if not rows:
break
# Iterate over the rows
for row in rows:
# Access the payload and update the corresponding qdrant point
card_html = row[8] if row[8] is not None else ""
tag_set = row[7] if row[7] is not None else ""
tracking_id = row[12] if row[12] is not None else ""
metadata = row[11] if row[11] is not None else ""
card = Card(card_html, tag_set, tracking_id, metadata)
card.send_post_request()
# Close the cursor and connection
cur.close()
conn.close()
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment