geodata-catalog/backend/src/crud.py

from typing import List

from sqlalchemy.orm import Session
from sqlalchemy.sql.expression import func
from sqlalchemy import inspect, text

from . import models, schemas
from .database import Base


def get_item(db: Session, item_id: int):
    """индивидуальные страницы для каждого Описания набора данных"""
    return db.query(models.Item).filter(models.Item.id == item_id).first()


def get_item_by_description(db: Session, needle: str, skip: int = 0, limit: int = 20):
    """фильтры для поиска Описаний наборов данных по атрибутам
    дополнительный возможный синтаксис в запросах преобразуется
    в поисковый запрос содержащий:
    - простой текст: переданные слова в любом порядке
    - "текст в кавычках": переданные слова в указанном порядке
    - OR: какие-либо из переданных слов
    - -: не содержащий данного слова
    """
    # hardcode russian for now
    # built-in postgress websearch_to_tsquery() is good
    # https://www.postgresql.org/docs/11/textsearch-controls.html#TEXTSEARCH-PARSING-QUERIES
    # TODO: add requested search columns [0]
    items_table = inspect(models.Item).local_table
    result = (
        db.query(models.Item)
        .filter(
            items_table.c.geodata_search_ts.op("@@")(
                func.websearch_to_tsquery('"russian"', needle)
            )
        )
        .order_by(items_table.c.id)
        .offset(skip)
        .limit(limit)
        .all()
    )
    return result


def get_item_by_detailed_description(
    db: Session, query: schemas.DetailedSearchQuery, skip: int = 0, limit: int = 20
):
    """фильтры для поиска Описаний наборов данных по атрибутам
    дополнительный возможный синтаксис основного запроса преобразуется
    в поисковый запрос содержащий:
    - простой текст: переданные слова в любом порядке
    - "текст в кавычках": переданные слова в указанном порядке
    - OR: какие-либо из переданных слов
    - -: не содержащий данного слова
    дополнительные поля ипользуются напрямую
    """
    # hardcode russian for now
    # built-in postgress websearch_to_tsquery() is good
    # https://www.postgresql.org/docs/11/textsearch-controls.html#TEXTSEARCH-PARSING-QUERIES
    items_table = inspect(models.Item).local_table
    result = db.query(models.Item)

    # filter with the main query (if present)
    if query.main_query is not None:
        result = result.filter(
            items_table.c.geodata_search_ts.op("@@")(
                func.websearch_to_tsquery('"russian"', query.main_query)
            )
        )
    # filter by additionalfields (if present)
    result = result.filter_by(**query.dict(exclude={"main_query"}, exclude_none=True))

    return result.order_by(items_table.c.id).offset(skip).limit(limit).all()


def get_items(db: Session, skip: int = 0, limit: int = 20):
    """список доступных в системе Описаний наборов данных"""
    return db.query(models.Item).offset(skip).limit(limit).all()


def get_items_by_category(category: str, db: Session, skip: int = 0, limit: int = 20):
    """список доступных в категории Описаний наборов данных"""
    return (
        db.query(models.Item)
        .filter(models.Item.category == category)
        .offset(skip)
        .limit(limit)
        .all()
    )


def insert_items(db: Session, items: List[schemas.ItemCreate]):
    """импорт в базу данных Описаний наборов данных"""
    before = db.query(models.Item).count()
    accepted = len(items)
    # https://docs.sqlalchemy.org/en/14/dialects/postgresql.html#psycopg2-fast-execution-helpers
    modeled_items = (models.ItemCreate(**item.dict()) for item in items)
    db.add_all(modeled_items)
    db.commit()
    inserted = db.query(models.Item).count() - before
    return (accepted, inserted)


def get_headers(db: Session):
    """полные заголовки таблиц"""
    return db.query(models.Header).all()


def add_spacial_data(db: Session):
    """Превращает строковые данные с координатами, полученные из таблицы
    в пространственные, и помещает их в отдельный столбец в общей базе данных
    """
    check_t = text(
        """SELECT true 
                FROM pg_attribute
                WHERE  attrelid = 'geodata'::regclass
                AND    attname = 'geom'
                AND    NOT attisdropped
                AND    attnum > 0"""
    )
    create_t = text(
        """SELECT AddGeometryColumn( 'public', 'geodata', 'geom', 4326, 'POINT', 2 )"""
    )
    populate_t = text(
        """UPDATE geodata
            SET geom = ST_SetSRID(ST_Point( y_coord::numeric, x_coord::numeric),4326)
        """
    )
    if not db.execute(check_t).all():
        db.execute(create_t)
    db.execute(populate_t)
    db.commit()