Попытка гать функции func для ошибок SQLite с помощью OperationalError sqlalchemy/sqlitePython

Программы на Python
Ответить Пред. темаСлед. тема
Anonymous
 Попытка гать функции func для ошибок SQLite с помощью OperationalError sqlalchemy/sqlite

Сообщение Anonymous »

Я пытаюсь написать пару пользовательских функций для SQLite в Sqlalchemy. Проблема, с которой я сталкиваюсь, заключается в том, что одна функция будет подана в другую, но ошибки выполнения, когда это произойдет, как если бы она не может передать возвращение одной функции другой.

Код: Выделить всё

class InString(GenericFunction):
type = String()  # noqa: A003
name = "INSTR"
identifier = "strpos"
inherit_cache = True

def ToChar(date, format: str) -> str:
convertedFormat = convertFormat(format)
if isinstance(date, int):
date = datetime.strptime(str(date), convertedFormat)
return date.strftime(convertedFormat)

def ToTimestamp(date: str, format: str):
convertedFormat = convertFormat(format)
return datetime.strptime(date, convertedFormat)

class UnitDatabase(Database):
def __init__(self, logger=logging.getLogger(__name__)):
metadata = Base.metadata
super().__init__("sqlite:///:memory:", logger)
self._session = Session(bind=self.engine)
for _, cur_table in metadata.tables.items():
for autoincrement_column in [
column for column in cur_table._columns if column.autoincrement is True
]:
if isinstance(autoincrement_column.type, BigInteger):
autoincrement_column.type = Integer()
if _.split(".")[0] not in cur_table.name:
cur_table.name = f"{cur_table.name}_{cur_table.schema}"
cur_table.schema = None
Base.metadata.create_all(self.engine)

@listens_for(Pool, 'connect')
def register_sqlite_functions(dbapi_conn, connection_record):
dbapi_conn.create_function("to_char", 2, ToChar)
dbapi_conn.create_function("to_timestamp", 2, ToTimestamp)

@property
def session(self):
return self._session

def convertFormat(format: str):
for key, value in FORMAT_DICT.items():
format = format.replace(key, value)
return format
< /code>
Нам нужна версия Tochres для работы для работы для SQLite, поэтому я написал методы, чтобы делать то, что нужно. Но проблема возникает, когда я использую их в заявлении. У меня есть TO_TIMESTAMP внутри TO_CHAR, TO TIMESTAMP проходит, когда я отлаживает, а значение выглядит точно так же, как и ожидалось. Но когда пришло время запустить TO_CHAR, это где -то ошибки (я не могу найти, где точно, это не в самих фактических функциях). Ошибка: < /p>
(sqlite3.OperationalError) user-defined function raised exception
< /code>
Вот упрощенный запрос: < /p>
worker_stmt = (
select()
.select_from(obs, sfc, lim)
.where(
obs.observationid == sfc.observationid,
obs.observationid == lim.observationid,
func.to_char(func.to_timestamp(cast(SfcObsTable.month, String), "MM"), "MM") == func.to_char(func.to_timestamp(cast(obs.observationtime, String), "YYYYMMDDHH24MISS"), "MM"),
)
)
< /code>
Ошибка является самой последней строкой с функциями. У меня есть небольшая потеря, я мог бы попытаться найти способ выполнить всю эту 1 функцию, но это может вызвать проблемы с другими утверждениями в других классах, в которых используется только один или другой. Есть ли что -то в моем коде, вызывая проблему? Или как функции вызываются в утверждении? Или какой -нибудь способ попытаться отладить это? Я попробовал < /p>
sqlite3.enable_callback_tracebacks(True)
Но это не предложило никаких дополнительных трассов, чтобы помочь или другие сообщения об ошибках, чтобы объяснить, что происходит.

Подробнее здесь: https://stackoverflow.com/questions/797 ... ror-sqlalc
Реклама
Ответить Пред. темаСлед. тема

Быстрый ответ

Изменение регистра текста: 
Смайлики
:) :( :oops: :roll: :wink: :muza: :clever: :sorry: :angel: :read: *x)
Ещё смайлики…
   
К этому ответу прикреплено по крайней мере одно вложение.

Если вы не хотите добавлять вложения, оставьте поля пустыми.

Максимально разрешённый размер вложения: 15 МБ.

  • Похожие темы
    Ответы
    Просмотры
    Последнее сообщение

Вернуться в «Python»