extracted client

filevine_client.py

@@ -0,0 +1,132 @@
+import os
+import requests
+from typing import List, Dict, Any, Optional
+
+# Load environment variables
+FV_CLIENT_ID = os.environ.get("FILEVINE_CLIENT_ID")
+FV_CLIENT_SECRET = os.environ.get("FILEVINE_CLIENT_SECRET")
+FV_PAT = os.environ.get("FILEVINE_PERSONAL_ACCESS_TOKEN")
+FV_ORG_ID = os.environ.get("FILEVINE_ORG_ID")
+FV_USER_ID = os.environ.get("FILEVINE_USER_ID")
+
+class FilevineClient:
+    def __init__(self, bearer_token: str = None):
+        self.bearer_token = bearer_token
+        self.base_url = "https://api.filevineapp.com/fv-app/v2"
+        self.headers = {
+            "Accept": "application/json",
+            "Authorization": f"Bearer {self.bearer_token}",
+            "x-fv-orgid": str(FV_ORG_ID),
+            "x-fv-userid": str(FV_USER_ID),
+        }
+    
+    def get_bearer_token(self) -> str:
+        """Get a new bearer token using Filevine credentials"""
+        url = "https://identity.filevine.com/connect/token"
+        data = {
+            "client_id": FV_CLIENT_ID,
+            "client_secret": FV_CLIENT_SECRET,
+            "grant_type": "personal_access_token",
+            "scope": "fv.api.gateway.access tenant filevine.v2.api.* email openid fv.auth.tenant.read",
+            "token": FV_PAT,
+        }
+
+        headers = {"Accept": "application/json"}
+        print(data)
+        resp = requests.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.headers["Authorization"] = f"Bearer {token}"
+        return token
+    
+    def list_all_projects(self) -> List[Dict[str, Any]]:
+        """Fetch all projects from Filevine API"""
+        base = f"{self.base_url}/Projects?limit=500"
+        results = []
+        last_count = None
+        tries = 0
+        offset = 0
+        cnt = 0
+        
+        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 = {}
+            if last_count is not None:
+                offset = offset + last_count
+                params["offset"] = offset
+            r = requests.get(url, headers=self.headers, params=params, timeout=30)
+            r.raise_for_status()
+            page = r.json()
+            items = page.get("items", [])
+            results.extend(items)
+            has_more = page.get("hasMore")
+            last_count = page.get("count")
+            if not has_more:
+                break
+            # Safety valve
+            if tries > 200:
+                break
+        return results
+    
+    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.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.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.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.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.raise_for_status()
+        return r.json().get("items")
+    
+    def fetch_form(self, project_id_native: int, form: str) -> Dict[str, Any]:
+        """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.raise_for_status()
+            return r.json()
+        except Exception as e:
+            print(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.raise_for_status()
+            return [x.get('dataObject') for x in r.json().get("items")]
+        except Exception as e:
+            print(e)
+            return {}