use-pypgstac

use-pypgstac.py

"""Utilities to bulk load data into pgstac from json/ndjson."""
import logging

from pypgstac.load import Loader as BaseLoader
logger = logging.getLogger(__name__)
from typing import Dict
from pypgstac.load import Methods
from stac_pydantic import Collection
from pypgstac.db import PgstacDB

class StacCollection(Collection):
    id: str
    item_assets: Dict


class Loader(BaseLoader):
    """Utilities for loading data and updating collection summaries/extents."""

    def __init__(self, db) -> None:
        super().__init__(db)
        self.check_version()
        self.conn = self.db.connect()
      
if __name__ == '__main__':
    DSN_STRING = 'postgresql://username:password@localhost:5439/postgis'
    FILE = '/Users/emiletenezakis/devseed/maap/veda-data-pipelines/data/collections/ESACCI_Biomass_L4_AGB_V3_100m_2018.json'
    import json
    with open(FILE, 'r') as f:
        coll_data = json.load(f)
    
    coll_data = StacCollection(**coll_data)
    try:
        with PgstacDB(dsn=DSN_STRING, debug=True) as db:
            loader = Loader(db=db)
            loader.load_collections(file=coll_data, insert_mode=Methods.upsert)
    except Exception as e:
        print(f"Encountered failure loading collection into pgSTAC: {e}")