Update app.py

app.py

@@ -1,117 +1,106 @@
-import logging
-from typing import Optional
-
-from flask import Flask, jsonify, request
+from fastapi import FastAPI, Query, HTTPException
+from pydantic import BaseModel
 from webscout import WEBS
 
-app = Flask(__name__)
-
-TIMEOUT = 10
-PROXY = None
-
-@app.route('/api/search', methods=['GET'])
-def search_text():
-    query = request.args.get('q', '')
-    max_results = request.args.get('max_results', 10, type=int)
-    timelimit = request.args.get('timelimit', None)
-    safesearch = request.args.get('safesearch', 'moderate')
-    region = request.args.get('region', 'wt-wt')
-
+app = FastAPI()
+
+class SearchQuery(BaseModel):
+    q: str
+    max_results: int = 10
+    timelimit: Optional[int] = None
+    safesearch: str = 'moderate'
+    region: str = 'wt-wt'
+
+class ImageSearchQuery(BaseModel):
+    q: str
+    max_results: int = 10
+    safesearch: str = 'moderate'
+    region: str = 'wt-wt'
+
+class VideoSearchQuery(BaseModel):
+    q: str
+    max_results: int = 10
+    safesearch: str = 'moderate'
+    region: str = 'wt-wt'
+    timelimit: Optional[int] = None
+    resolution: Optional[str] = None
+    duration: Optional[str] = None
+
+class NewsSearchQuery(BaseModel):
+    q: str
+    max_results: int = 10
+    safesearch: str = 'moderate'
+    region: str = 'wt-wt'
+    timelimit: Optional[int] = None
+
+class MapSearchQuery(BaseModel):
+    q: str
+    place: Optional[str] = None
+    max_results: int = 10
+
+class TranslateQuery(BaseModel):
+    q: str
+    to: str = 'en'
+
+class SuggestionQuery(BaseModel):
+    q: str
+
+@app.get("/api/search", response_model=List[dict])
+async def search_text(query: SearchQuery):
     results = []
     with WEBS() as webs:
-        for result in webs.text(query, max_results=max_results, timelimit=timelimit, safesearch=safesearch, region=region):
+        for result in webs.text(**query.dict()):
             results.append(result)
+    return results
 
-    return jsonify({'results': results})
-
-
-@app.route('/api/images', methods=['GET'])
-def search_images():
-    query = request.args.get('q', '')
-    max_results = request.args.get('max_results', 10, type=int)
-    safesearch = request.args.get('safesearch', 'moderate')
-    region = request.args.get('region', 'wt-wt')
+@app.get("/api/images", response_model=List[dict])
+async def search_images(query: ImageSearchQuery):
     results = []
     with WEBS() as webs:
-        for result in webs.images(query, max_results=max_results, safesearch=safesearch, region=region):
+        for result in webs.images(**query.dict()):
             results.append(result)
+    return results
 
-    return jsonify({'results': results})
-
-@app.route('/api/videos', methods=['GET'])
-def search_videos():
-    query = request.args.get('q', '')
-    max_results = request.args.get('max_results', 10, type=int)
-    safesearch = request.args.get('safesearch', 'moderate')
-    region = request.args.get('region', 'wt-wt')
-    timelimit = request.args.get('timelimit', None)
-    resolution = request.args.get('resolution', None)
-    duration = request.args.get('duration', None)
-
+@app.get("/api/videos", response_model=List[dict])
+async def search_videos(query: VideoSearchQuery):
     results = []
     with WEBS() as webs:
-        for result in webs.videos(query, max_results=max_results, safesearch=safesearch, region=region, timelimit=timelimit, resolution=resolution, duration=duration):
+        for result in webs.videos(**query.dict()):
             results.append(result)
+    return results
 
-    return jsonify({'results': results})
-
-@app.route('/api/news', methods=['GET'])
-def search_news():
-    query = request.args.get('q', '')
-    max_results = request.args.get('max_results', 10, type=int)
-    safesearch = request.args.get('safesearch', 'moderate')
-    region = request.args.get('region', 'wt-wt')
-    timelimit = request.args.get('timelimit', None)
-
+@app.get("/api/news", response_model=List[dict])
+async def search_news(query: NewsSearchQuery):
     results = []
     with WEBS() as webs:
-        for result in webs.news(query, max_results=max_results, safesearch=safesearch, region=region, timelimit=timelimit):
+        for result in webs.news(**query.dict()):
             results.append(result)
+    return results
 
-    return jsonify({'results': results})
-
-@app.route('/api/maps', methods=['GET'])
-def search_maps():
-    query = request.args.get('q', '')
-    place = request.args.get('place', None)
-    max_results = request.args.get('max_results', 10, type=int)
-
+@app.get("/api/maps", response_model=List[dict])
+async def search_maps(query: MapSearchQuery):
     results = []
     with WEBS() as webs:
-        for result in webs.maps(query, place=place, max_results=max_results):
+        for result in webs.maps(**query.dict()):
             results.append(result)
+    return results
 
-    return jsonify({'results': results})
-
-@app.route('/api/translate', methods=['GET'])
-def translate_text():
-    query = request.args.get('q', '')
-    to_lang = request.args.get('to', 'en')
-
+@app.get("/api/translate", response_model=dict)
+async def translate_text(query: TranslateQuery):
     with WEBS() as webs:
-        translation = webs.translate(query, to=to_lang)
-
-    return jsonify({'translation': translation})
-
-@app.route('/api/suggestions', methods=['GET'])
-def search_suggestions():
-    query = request.args.get('q', '')
-    if not query:
-        return jsonify({'error': 'Query parameter missing'})
+        translation = webs.translate(**query.dict())
+    return {'translation': translation}
 
+@app.get("/api/suggestions", response_model=List[str])
+async def search_suggestions(query: SuggestionQuery):
+    if not query.q:
+        raise HTTPException(status_code=400, detail="Query parameter missing")
     results = []
-    try:
-        with WEBS() as webs:
-            for result in webs.suggestions(query):
-                results.append(result)
-    except Exception as e:
-        return jsonify({'error': str(e)}), 500
-
-    return jsonify({'results': results})
-
-@app.route('/api/health', methods=['GET'])
-def health_check():
-    return jsonify({'status': 'working'})
+    with WEBS() as webs:
+        for result in webs.suggestions(query.q):
+            results.append(result)
+    return results
 
-if __name__ == '__main__':
-    app.run(debug=True)
+@app.get("/api/health", response_model=dict)
+async def health_check():
+    return {'status': 'working'}