mirror of
ssh://git.janware.com/srv/git/janware/proj/jw-python
synced 2026-01-15 01:52:56 +01:00
93 lines
2.6 KiB
Python
93 lines
2.6 KiB
Python
# -*- coding: utf-8 -*-
|
|
|
|
from typing import Any
|
|
|
|
import abc
|
|
|
|
from jwutils.log import *
|
|
from jwutils.misc import load_classes
|
|
from jwutils.Cmds import Cmds
|
|
from jwutils.db.DataBase import DataBase
|
|
from jwutils.db.query.Query import Query as QueryBase
|
|
from jwutils.db.query.QueryResult import QueryResult
|
|
from jwutils.db.schema.Schema import Schema
|
|
|
|
class Queries(abc.ABC): # export
|
|
|
|
class Query(QueryBase):
|
|
|
|
def __init__(self: Any, query: QueryBase, name: str, location: str, func: Any):
|
|
self.__query = query
|
|
self.__func = func
|
|
self.__location = location
|
|
self.__name = name
|
|
|
|
# -- implement API
|
|
|
|
def _run(self, *args, **kwargs) -> QueryResult:
|
|
return self.__func(*args, **kwargs)
|
|
|
|
def _register(self):
|
|
raise Exception('Can\'t call register on this object')
|
|
|
|
def _column_names(self) -> list[str]:
|
|
return self.__query.column_names
|
|
|
|
@property
|
|
def name(self):
|
|
return self.__name
|
|
|
|
def __init__(self, db: DataBase) -> None:
|
|
assert(isinstance(db, DataBase))
|
|
self.__db = db
|
|
self.__queries: dict[str, Any] = dict()
|
|
|
|
def __getitem__(self, name) -> Query:
|
|
try:
|
|
return self.__queries[name]
|
|
except Exception as e:
|
|
slog(ERR, f'Failed to retrieve query {name} ({e})')
|
|
slog_m(ERR, '\n'.join(self.__queries.keys()))
|
|
raise
|
|
|
|
@property
|
|
def db(self) -> DataBase:
|
|
return self.__db
|
|
|
|
@property
|
|
def app(self) -> Cmds:
|
|
return self.__db.app
|
|
|
|
def load(self, modules: list[str], cls=QueryBase):
|
|
for path in modules:
|
|
slog(INFO, f'Loading modules from {path}')
|
|
for c in load_classes(path, cls):
|
|
c(self).register()
|
|
|
|
@property
|
|
def schema(self) -> Schema:
|
|
return self.__db.schema
|
|
|
|
def add(self, query: QueryBase, query_name: str, location: str, func: Any):
|
|
slog(INFO, f'Adding query "{query_name}" on location "{location}"')
|
|
assert(isinstance(query_name, str))
|
|
assert(isinstance(location, str))
|
|
#ret = self.Query(query, func)
|
|
ret = self.Query(query, query_name, location, func)
|
|
#setattr(ret, 'name', name)
|
|
self.__queries[query_name] = ret
|
|
self._add(ret)
|
|
|
|
# To be overriden in case the derived class wants to keep track
|
|
def _add(self, query):
|
|
pass
|
|
|
|
# def register(self):
|
|
# return self._register()
|
|
#
|
|
# def query(self, *args, **kwargs):
|
|
# return self.app.query(*args, **kwargs)
|
|
#
|
|
# def pages(self, *args, **kwargs):
|
|
# return self.app.pages(*args, **kwargs)
|
|
#
|