fastapi server

README.md

@@ -2,3 +2,6 @@ wget https://dumps.wikimedia.org/simplewiki/20250420/simplewiki-20250420-pages-a
 
 
 python db/wiki_parser_sqlite.py simplewiki-20250420-pages-articles-multistream.xml.bz2 db/data/wikihop.db --batch-size 10000
+
+
+WIKI_DB_PATH=db/data/wikihop.db python api_server.py 

api_server.py

@@ -0,0 +1,68 @@
+from fastapi import FastAPI, HTTPException, Query
+from fastapi.middleware.cors import CORSMiddleware
+import uvicorn
+import os
+from typing import List, Dict, Any, Optional
+from db.wiki_db_sqlite import WikiDBSqlite
+
+app = FastAPI(title="Wiki API Server")
+
+# Add CORS middleware
+app.add_middleware(
+    CORSMiddleware,
+    allow_origins=["*"],
+    allow_credentials=True,
+    allow_methods=["*"],
+    allow_headers=["*"],
+)
+
+# Global database connection
+db = None
+
+@app.on_event("startup")
+async def startup_db_client():
+    global db
+    db_path = os.environ.get("WIKI_DB_PATH")
+    if not db_path:
+        raise ValueError("WIKI_DB_PATH environment variable not set")
+    db = WikiDBSqlite(db_path)
+
+@app.get("/article_count")
+async def get_article_count() -> Dict[str, int]:
+    """Get the number of articles in the database"""
+    return {"count": db.get_article_count()}
+
+@app.get("/article_titles")
+async def get_article_titles() -> Dict[str, List[str]]:
+    """Get all article titles"""
+    return {"titles": db.get_all_article_titles()}
+
+@app.get("/article")
+async def get_article(title: str = Query(..., description="Article title")) -> Dict[str, Any]:
+    """Get article data by title"""
+    article = db.get_article(title)
+    if not article:
+        raise HTTPException(status_code=404, detail=f"Article '{title}' not found")
+    return article
+
+@app.get("/article_exists")
+async def article_exists(title: str = Query(..., description="Article title")) -> Dict[str, bool]:
+    """Check if an article exists"""
+    return {"exists": db.article_exists(title)}
+
+@app.get("/article_text")
+async def get_article_text(title: str = Query(..., description="Article title")) -> Dict[str, str]:
+    """Get the text of an article"""
+    if not db.article_exists(title):
+        raise HTTPException(status_code=404, detail=f"Article '{title}' not found")
+    return {"text": db.get_article_text(title)}
+
+@app.get("/article_links")
+async def get_article_links(title: str = Query(..., description="Article title")) -> Dict[str, List[str]]:
+    """Get the links of an article"""
+    if not db.article_exists(title):
+        raise HTTPException(status_code=404, detail=f"Article '{title}' not found")
+    return {"links": db.get_article_links(title)}
+
+if __name__ == "__main__":
+    uvicorn.run("api_server:app", host="0.0.0.0", port=8000, reload=True) 

db/wiki_db_api.py

@@ -0,0 +1,60 @@
+import requests
+
+class WikiDBAPI:
+    def __init__(self, api_endpoint):
+        """Initialize the database with API endpoint URL"""
+        self.api_endpoint = api_endpoint.rstrip('/')
+        # Verify connection and get initial article count
+        self._article_count = self._get_article_count()
+        print(f"Connected to Wiki API at {api_endpoint} with {self._article_count} articles")
+    
+    def __del__(self):
+        """Clean up resources when object is destroyed"""
+        # No persistent connection to close in the API version
+        pass
+    
+    def _get_article_count(self):
+        """Get the number of articles via API"""
+        response = requests.get(f"{self.api_endpoint}/article_count")
+        response.raise_for_status()
+        return response.json()["count"]
+    
+    def get_article_count(self):
+        """Return the number of articles"""
+        return self._article_count
+        
+    def get_all_article_titles(self):
+        """Return a list of all article titles"""
+        response = requests.get(f"{self.api_endpoint}/article_titles")
+        response.raise_for_status()
+        return response.json()["titles"]
+        
+    def get_article(self, title):
+        """Get article data by title"""
+        response = requests.get(f"{self.api_endpoint}/article", params={"title": title})
+        if response.status_code == 404:
+            return {}
+        response.raise_for_status()
+        return response.json()
+        
+    def article_exists(self, title):
+        """Check if an article exists"""
+        response = requests.get(f"{self.api_endpoint}/article_exists", params={"title": title})
+        response.raise_for_status()
+        return response.json()["exists"]
+        
+    def get_article_text(self, title):
+        """Get the text of an article"""
+        response = requests.get(f"{self.api_endpoint}/article_text", params={"title": title})
+        if response.status_code == 404:
+            return ''
+        response.raise_for_status()
+        return response.json()["text"]
+        
+    def get_article_links(self, title):
+        """Get the links of an article"""
+        response = requests.get(f"{self.api_endpoint}/article_links", params={"title": title})
+        if response.status_code == 404:
+            return []
+        response.raise_for_status()
+        return response.json()["links"]

engine.py

@@ -6,14 +6,18 @@
 import random
 from db.wiki_db_sqlite import WikiDBSqlite
 from db.wiki_db_json import WikiDBJson
-
+from db.wiki_db_api import WikiDBAPI
 class WikiRunEnvironment:
     def __init__(self, wiki_data_path):
         """Initialize with path to Wikipedia data"""
         if wiki_data_path.endswith('.json'):
             self.db = WikiDBJson(wiki_data_path)
-        else:
+        elif wiki_data_path.endswith('.db'):
             self.db = WikiDBSqlite(wiki_data_path)
+        elif wiki_data_path.startswith('http'):
+            self.db = WikiDBAPI(wiki_data_path)
+        else:
+            raise ValueError("Invalid file extension. Must be .json or .db")
 
         self.current_article = None
         self.target_article = None