Merge pull request #3 from dvrpc/colin-branch

Changed content/viz repository to postgres db

Author: ckirby98

scripts/convert_github_content_to_db.py

@@ -0,0 +1,185 @@
+
+import psycopg
+import requests
+import os
+import logging
+import asyncio
+import aiohttp
+from dotenv import load_dotenv
+from fastapi_cache.decorator import cache
+import pandas as pd
+from sqlalchemy import create_engine
+from datetime import datetime
+
+# from utils.consts import subcategory_map
+
+log = logging.getLogger(__name__)
+
+load_dotenv()
+token = os.getenv("GITHUB_TOKEN")
+headers = {
+    "Authorization": f"Bearer {token}"
+}
+base_contents_url = "https://api.github.com/repos/dvrpc/community-profiles-content/contents"
+
+subcategory_map = {
+    'demographics-housing': {'demographics': [], 'housing': []},
+    'economy': {'employment': [], 'income-poverty': [], 'transportation': []},
+    'active-transportation': {'cycling': [], 'pedestrian': [], 'commute': []},
+    'safety-health': {'crash': [], 'health': []},
+    'freight': {'freight': []},
+    'environment': {'open-space': [], 'planning': []},
+    'transit': {'transit': [], 'tip': []},
+    'roadways': {'conditions': [], 'tip': []}
+}
+
+host = os.getenv("DB_HOST")
+dbname = os.getenv("DB_NAME")
+user = os.getenv("DB_USER")
+password = os.getenv("DB_PASS")
+port = os.getenv("DB_PORT")
+
+uri = f"postgresql://{user}:{password}@{host}:{port}/{dbname}"
+print(uri)
+# establish connection with the database
+engine = create_engine(uri)
+
+async def get_viz_download_url(geo_level, category, subcategory, topic):
+    url = f"{base_contents_url}/{geo_level}/viz/{category}/{subcategory}/{topic}.json"
+    try:
+        r = requests.get(url, headers=headers)
+        r.raise_for_status()
+
+    except requests.exceptions.HTTPError as e:
+        log.error(
+            f"Error fetching {geo_level} contents: \n{e}")
+
+    data =  r.json()
+
+    return data['download_url']
+
+
+async def get_download_urls(geo_level, type):
+    urls = []
+    
+    for key, value in subcategory_map.items():
+        for subcategory in value.keys():
+            urls.append({
+                'category': key,
+                'subcategory': subcategory,
+                'geo_level': geo_level,
+                'url': f"{base_contents_url}/{geo_level}/{type}/{key}/{subcategory}" 
+            })
+    
+    
+    async with aiohttp.ClientSession() as session:
+        tasks = []
+        for url in urls:
+            tasks.append(get_download_url(session, url, type))
+
+        results = await asyncio.gather(*tasks)
+        flattened = [item for sublist in results for item in sublist]
+        return flattened
+
+async def get_download_url(session, url, type):
+    try:
+        async with session.get(url['url'], headers=headers) as response:
+            if response.status == 200:
+                data = await response.json()
+                download_urls = []
+                for file in data:
+                    index = -5 if type == 'viz' else -3
+                    download_urls.append({
+                        'category': url['category'],
+                        'subcategory': url['subcategory'],
+                        'geo_level': url['geo_level'],
+                        'name': file['name'][:index],
+                        'url': file['download_url'],
+                    })
+                return download_urls
+            else:
+                log.info(response.status)
+    except aiohttp.ClientConnectionError as e:
+        log.error(f'Connection error: {e}')
+        
+async def get_files(urls):
+    async with aiohttp.ClientSession() as session:
+        tasks = []
+        for url in urls:
+            tasks.append(get_file(session, url))
+
+        results = await asyncio.gather(*tasks)
+        return results
+
+
+async def get_file(session, url):
+    try:
+        async with session.get(url['url']) as response:
+            if response.status == 200:
+                text = await response.text()
+                return {
+                    'category': url['category'],
+                    'subcategory': url['subcategory'],
+                    'geo_level': url['geo_level'],
+                    'name': url['name'],
+                    'file': text,
+                }
+            else:
+                log.info(response.status)
+    except aiohttp.ClientConnectionError as e:
+        log.error(f'Connection error: {e}')
+
+
+# def get_file(url):
+#     try:
+#         r = requests.get(url)
+#         r.raise_for_status()
+
+#     except requests.exceptions.HTTPError as e:
+#         log.error(
+#             f"Error fetching {url}: \n{e}")
+
+#     return r.json()
+
+async def save_content(geo_level):
+    all_urls = []
+    for geo_level in ['region', 'county', 'municipality']:
+        md_download_urls = await get_download_urls(geo_level, 'md')
+        all_urls += md_download_urls
+
+    files = await get_files(all_urls)
+    df = pd.DataFrame(files)
+    
+    current_local_time = datetime.now()
+    df['create_date'] = current_local_time
+    df.to_sql('content', con=engine, if_exists='replace',index=False)
+    
+async def save_visualizations(geo_level):
+    all_urls = []
+
+    for geo_level in ['region', 'county', 'municipality']:
+        viz_download_urls = await get_download_urls(geo_level, 'viz')
+        all_urls += viz_download_urls
+
+
+    files = await get_files(all_urls)
+    df = pd.DataFrame(files)
+    
+    current_local_time = datetime.now()
+    df['create_date'] = current_local_time
+    df.to_sql('visualizations', con=engine, if_exists='replace',index=False)
+    
+asyncio.run(save_content('county'))
+asyncio.run(save_visualizations('county'))
+
+
+
+
+# content_map = copy.deepcopy(subcategory_map)
+# for md in files:
+#     content_map[md['category']][md['subcategory']].append({
+#         'name': md['name'],
+#         'content': content
+#     })
+
+# return content_map

src/repository/__init__.py

@@ -3,10 +3,11 @@
 import logging
 import psycopg
 import asyncio
+import json
 log = logging.getLogger(__name__)
 
 
-def fetch_profile(query):
+def fetch_one(query):
     cur = db.conn.cursor()
     response = None
 
@@ -24,12 +25,33 @@ def fetch_profile(query):
 
     return response
 
+def fetch_many(query):
+    cur = db.conn.cursor()
+    response = None
+
+    try:
+        cur.execute(query)
+        data = cur.fetchall()
+        column_names = [desc[0] for desc in cur.description]
+        
+        response = []
+        for row in data:
+            response.append(dict(zip(column_names, row)))
+    except psycopg.OperationalError as err:
+        log.error(f"Connection exception executing: \n{query} \n{err}")
+    except psycopg.Error as err:
+        log.error(f"Other psycopg error executing: \n{query} \n{err}")
+    except Exception as err:
+        log.error(f"Error executing query: \n{query} \n{err}")
+
+    return response
+
 
 @cache(expire=60)
 async def fetch_county(geoid):
     log.info(f'Fetching county profile: {geoid}')
     query = f"SELECT * FROM county WHERE geoid = '{geoid}'"
-    profile = fetch_profile(query)
+    profile = fetch_one(query)
     log.info(f'Succesfully retrieved county profile: {geoid}')
     return profile
 
@@ -38,14 +60,72 @@ async def fetch_county(geoid):
 async def fetch_municipality(geoid):
     log.info(f'Fetching municipality profile: {geoid}')
     query = f"SELECT * FROM municipality WHERE geoid = '{geoid}'"
-    profile = fetch_profile(query)
+    profile = fetch_one(query)
     log.info(f'Succesfully retrieved municipality profile: {geoid}')
     return profile
 
 @cache(expire=60)
 async def fetch_region():
     log.info(f'Fetching regional profile')
     query = "SELECT * FROM region"
-    profile = fetch_profile(query)
+    profile = fetch_one(query)
     log.info(f'Succesfully retrieved regional profile')
     return profile
+
+@cache(expire=60)
+async def fetch_content(geo_level):
+    log.info(f'Fetching {geo_level} content...')
+    query = f"SELECT category, subcategory, name, file FROM content WHERE geo_level = '{geo_level}'"
+    response = fetch_many(query)
+    log.info(f'Succesfully retrieved {geo_level} content')
+    return response
+
+
+@cache(expire=60)
+async def fetch_visualizations(geo_level, category, subcategory, topic):
+    log.info(f'Fetching {geo_level}/{category}/{subcategory}/{topic} visualizations...')
+    query = f"""SELECT file FROM visualizations WHERE 
+                geo_level = '{geo_level}' 
+                AND category = '{category}' 
+                AND subcategory = '{subcategory}' 
+                AND name = '{topic}' """
+                
+    response = fetch_one(query)
+    log.info(f'Succesfully retrieved {geo_level}/{category}/{subcategory}/{topic} visualizations')
+    return json.loads(response['file'])
+
+async def fetch_content_template(geo_level: str, category: str, subcategory: str, topic: str):
+    log.info(f'Fetching {geo_level}/{category}/{subcategory}/{topic} content template...')
+    query = f"""SELECT file FROM content WHERE 
+            geo_level = '{geo_level}' 
+            AND category = '{category}' 
+            AND subcategory = '{subcategory}' 
+            AND name = '{topic}' """
+            
+    response = fetch_one(query)
+    log.info(f'Succesfully retrieved {geo_level}/{category}/{subcategory}/{topic} content template...')
+    return response['file']
+
+async def fetch_viz_template(geo_level: str, category: str, subcategory: str, topic: str):
+    log.info(f'Fetching {geo_level}/{category}/{subcategory}/{topic} viz template...')
+    query = f"""SELECT file FROM visualizations WHERE 
+            geo_level = '{geo_level}' 
+            AND category = '{category}' 
+            AND subcategory = '{subcategory}' 
+            AND name = '{topic}' """
+            
+    response = fetch_one(query)
+    log.info(f'Succesfully retrieved {geo_level}/{category}/{subcategory}/{topic} viz template...')
+    return json.loads(response['file'])
+
+async def fetch_template_tree(geo_level: str):
+    query = f"select category, subcategory, name from content where geo_level = '{geo_level}'"
+    response = fetch_many(query)
+    return response
+
+async def fetch_single_content(category: str, subcategory: str, topic: str):
+    log.info(f'Fetching  {category}/{subcategory}/{topic} content ...')
+    query = f"select file from content where category = '{category}' and subcategory = '{subcategory}' and name = '{topic}'"
+    response = fetch_one(query)
+    log.info(f'Succesfully fetched /{category}/{subcategory}/{topic} content')
+    return response['file']

src/repository/profile_repository.py

@@ -1,9 +1,8 @@
-from fastapi import APIRouter
-from repository.profile_repository import fetch_county, fetch_municipality, fetch_region
-from services.content import build_content
+from fastapi import APIRouter, Body
+from repository.profile_repository import fetch_content_template, fetch_county, fetch_municipality, fetch_region
+from services.content import build_content, build_single_content, build_template_tree
 from fastapi.responses import HTMLResponse
 from fastapi.staticfiles import StaticFiles
-from fastapi.templating import Jinja2Templates
 
 router = APIRouter(
     prefix="/content",
@@ -14,7 +13,6 @@
 #     profile = build_tract_profile(geoid)
 #     return profile
 
-templates = Jinja2Templates(directory="content")
 
 
 @router.get("/municipality/{geoid}")
@@ -35,4 +33,21 @@ async def get_county(geoid: str):
 async def get_region():
     profile = await fetch_region()
     content = await build_content('region', profile)
-    return content
+    return content
+
+@router.post('/preview')
+async def get_content_template(category: str, subcategory: str, topic: str, body: str = Body(..., media_type="text/plain")):
+    profile = await fetch_region()
+
+    template = await build_single_content(body, profile,category, subcategory, topic)
+    return template
+
+@router.get('/template/{geo_level}')
+async def get_content_template(geo_level: str, category: str, subcategory: str, topic: str):
+    template = await fetch_content_template(geo_level, category, subcategory, topic)
+    return template
+
+@router.get('/template/tree/{geo_level}')
+async def get_template_tree(geo_level: str):
+    tree = await build_template_tree(geo_level)
+    return tree

src/routers/content.py

@@ -1,5 +1,5 @@
 from fastapi import APIRouter
-from repository.profile_repository import fetch_county, fetch_municipality, fetch_region
+from repository.profile_repository import fetch_county, fetch_municipality, fetch_region, fetch_viz_template
 from services.viz import build_visualizations
 from fastapi_cache.decorator import cache
 
@@ -25,4 +25,9 @@ async def get_municipality_visualizations(geoid: str, category: str, subcategory
 async def get_region_visualization(category: str, subcategory: str, topic: str):
     profile = await fetch_region()
     viz = await build_visualizations('region', profile, category, subcategory, topic)
-    return viz
+    return viz
+
+@router.get('/template/{geo_level}')
+async def get_viz_template(geo_level: str, category: str, subcategory: str, topic: str):
+    template = await fetch_viz_template(geo_level, category, subcategory, topic)
+    return template