SqlAlchemy Oracle DataTime格式

Question

我使用SqlAlchemy查询Oracle数据库并将结果存储在csv文件中。

我想为要写入的日期指定一种全局格式，如下所示：

'DD-MM-YYYY HH24:MI:SS'.

我已经在系统上设置了NLS_DATE_FORMAT。

例如：

datetime.datetime(2016, 12, 22, 13, 12, 35)

最终将：

2004-12-22 13:12:35

我想：

22-12-2004 13:12:35

当我处理数百个表时，我无法“手动”应用“strftime”。

Answer 1

也许我的目的和你一样。

我只想打印（行）已格式化日期时间，而不是数据时间。日期时间（y、m、d…事物。

我尝试了几种方法（关于处理datetime）：

• sql level convert使用funcdate\u format//与sql方言相关的
从设备中选择*，date\u format（上次登录，“%m-%d-%Y”）
//函数。date\u格式，func。至\u char
• 将RowProxy转换为普通dict//不同的访问样式
• 自定义JSON。编码器//仅用于json字符串化
https://stackoverflow.com/a/22238613/4896468
• 继承RowProxy//我认为入侵性较小

只需在这里解释继承方式：

原始行

>>> type(row)
<class 'aiomysql.sa.result.RowProxy'>

>>> row
(4982, '203233D81914', False, datetime.datetime(2021, 6, 28, 19, 30, 6))

>>> row.last_login
datetime.datetime(2021, 6, 28, 19, 30, 6)

我那一排

>>> my_row = my_proxy(row)
>>> my_row
(4982, '203233D81914', False, '2021-06-28 19:30:06')

>>> my_row.last_login
'2021-06-28 19:30:06'

继承RowProxy并重写（根据需要任意重写）

• __repr__（用于打印）
• __getattr__（用于. xxx文件访问）

from datetime import datetime

class my_proxy(proxy_type):
  # refer orig RowProxy implement
  def __init__(self, row):
    self._result_proxy = row._result_proxy
    self._row = row._row
    self._processors = row._processors
    self._keymap = row._keymap
  # for print
  def __repr__(self): 
    # return ' xx:'+ repr(self.as_tuple())
    return repr(tuple(v.strftime('%F %T') if type(v)==datetime else v for v in self.as_tuple()))
  # for field access
  def __getattr__(self, name):
    try:
      # return self[name]
      v = self[name]
      return v.strftime('%F %T') if type(v)==datetime else v
    except KeyError as e:
      raise AttributeError(e.args[0])

proxy_type=类型（行）

这是aiomysql.sa.result.RowProxy

代码基于

• aiomysql 0.0.22
• SQLAlchemy 1.4.31

尤其是\uuu init\uuuu函数：
　github:aimysql/sa/result。py#L22

不同的版本可能会有所不同。

Answer 2

我找到了解决这个问题的方法。

是的，可以将日期转换为具有适当格式的字符。但是，在我的例子中，SQL语句是由另一个模块提供的，我需要处理100多个表。

因此，我决定使用由SqlAlchemy通过执行（）方法返回的ResultProxy对象中包含的数据。

我一次按1000行的块（块是经典类型列表）获取一个表。但这些行是一种元组（更准确地说是SqlAlchemy RowProxy对象），它无法修改。

因此，我对它们进行了处理，将它们按顺序排列，并更新组块列表。

使用“集合”很重要。OrderedDict’因为它保持字段的顺序。对于经典dic，则字段标签和值可能不匹配。

现在，我的区块可以进行各种处理（将日期更改为具有适当格式的字符串，在VARCHAR字符串中替换char，等等……）。字典结构非常适合这一点。

请注意，在写入之前，必须回溯区块列表中的OrderedDic行。

result_proxy = connection.execute(request)
while True:
    chunk = self.result_proxy.fetchmany(1000)
    if not chunk:
        break
    # treatments comes here after :

    # 1- transform into a dic in order to be able to modify
    for i, row in enumerate(chunk):
        chunk[i] = OrderedDict(row)

    # 2- clean dates
    for i, row_odic in enumerate(chunk):
        for item in row_odic:
            if(type(row_odic[item]) is datetime.datetime):
                row_odic[item] = str(row_odic[item].strftime("%d/%m/%Y"))
        chunk[i] = row_odic

    # Other data treatment

    # cast back for it to look like a classical result :
    for c, row_odic in enumerate(chunk):
        self.chunk[c] = row_odic.values()

    # finally write row_odic.values in the csv file

我不确定这是否是最有效的解决方案，但性能看起来不错。我有一个版本的这种处理（相同数量的数据），但使用Pandas库，执行起来有点长。