notid/rothbard
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

refactor: use requests Session for connection pooling and clean up debug output

Browse Source 
Replace per-request requests.get/post with a shared Session for connection reuse.

Remove verbose print statements and add structured [WARN] prefixes to error logs.
This commit is contained in:
Bryce
2026-05-12 23:38:46 -07:00
parent dc81c8e2a7
commit 9df9e003c1
1 changed files with 18 additions and 15 deletions
Download Patch File Download Diff File Expand all files Collapse all files

33
filevine_client.py
View File

@@ -19,6 +19,12 @@ class FilevineClient:
"x-fv-orgid": str(FV_ORG_ID),
"x-fv-userid": str(FV_USER_ID),
}
self.session = requests.Session()
self.session.headers.update({
"Accept": "application/json",
"x-fv-orgid": str(FV_ORG_ID),
"x-fv-userid": str(FV_USER_ID),
})
self.get_bearer_token()
def get_bearer_token(self) -> str:
@@ -33,14 +39,12 @@ class FilevineClient:
}
headers = {"Accept": "application/json"}
print("data is", data)
print(data)
resp = requests.post(url, data=data, headers=headers, timeout=30)
resp = self.session.post(url, data=data, headers=headers, timeout=30)
resp.raise_for_status()
js = resp.json()
token = js.get("access_token")
print(f"Got bearer js", js)
self.bearer_token = token
self.session.headers["Authorization"] = f"Bearer {token}"
self.headers["Authorization"] = f"Bearer {token}"
return token
@@ -60,7 +64,6 @@ class FilevineClient:
while True:
cnt = len(results)
print(f"list try {tries}, starting at {offset}, previous count {last_count}, currently at {cnt}")
tries += 1
url = base
params = {}
@@ -72,7 +75,7 @@ class FilevineClient:
if latest_activity_since:
params["latestActivitySince"] = latest_activity_since
r = requests.get(url, headers=self.headers, params=params, timeout=30)
r = self.session.get(url, headers=self.headers, params=params, timeout=30)
r.raise_for_status()
page = r.json()
items = page.get("items", [])
@@ -89,35 +92,35 @@ class FilevineClient:
def fetch_project_detail(self, project_id_native: int) -> Dict[str, Any]:
"""Fetch detailed information for a specific project"""
url = f"{self.base_url}/Projects/{project_id_native}"
r = requests.get(url, headers=self.headers, timeout=30)
r = self.session.get(url, headers=self.headers, timeout=30)
r.raise_for_status()
return r.json()
def fetch_project_team(self, project_id_native: int) -> List[Dict[str, Any]]:
"""Fetch team members for a specific project"""
url = f"{self.base_url}/Projects/{project_id_native}/team?limit=1000"
r = requests.get(url, headers=self.headers, timeout=30)
r = self.session.get(url, headers=self.headers, timeout=30)
r.raise_for_status()
return r.json().get('items') or []
def fetch_project_tasks(self, project_id_native: int) -> Dict[str, Any]:
"""Fetch tasks for a specific project"""
url = f"{self.base_url}/Projects/{project_id_native}/tasks"
r = requests.get(url, headers=self.headers, timeout=30)
r = self.session.get(url, headers=self.headers, timeout=30)
r.raise_for_status()
return r.json()
def fetch_client(self, client_id_native: int) -> Dict[str, Any]:
"""Fetch client information by client ID"""
url = f"{self.base_url}/contacts/{client_id_native}"
r = requests.get(url, headers=self.headers, timeout=30)
r = self.session.get(url, headers=self.headers, timeout=30)
r.raise_for_status()
return r.json()
def fetch_contacts(self, project_id_native: int) -> Optional[List[Dict[str, Any]]]:
"""Fetch contacts for a specific project"""
url = f"{self.base_url}/projects/{project_id_native}/contacts"
r = requests.get(url, headers=self.headers, timeout=30)
r = self.session.get(url, headers=self.headers, timeout=30)
r.raise_for_status()
return r.json().get("items")
@@ -125,20 +128,20 @@ class FilevineClient:
"""Fetch a specific form for a project"""
try:
url = f"{self.base_url}/Projects/{project_id_native}/Forms/{form}"
r = requests.get(url, headers=self.headers, timeout=30)
r = self.session.get(url, headers=self.headers, timeout=30)
r.raise_for_status()
return r.json()
except Exception as e:
print(e)
print(f"[WARN] Failed to fetch form '{form}' for project {project_id_native}: {e}")
return {}
def fetch_collection(self, project_id_native: int, collection: str) -> List[Dict[str, Any]]:
"""Fetch a collection for a project"""
try:
url = f"{self.base_url}/Projects/{project_id_native}/Collections/{collection}"
r = requests.get(url, headers=self.headers, timeout=30)
r = self.session.get(url, headers=self.headers, timeout=30)
r.raise_for_status()
return [x.get('dataObject') for x in r.json().get("items")]
except Exception as e:
print(e)
print(f"[WARN] Failed to fetch collection '{collection}' for project {project_id_native}: {e}")
return {}