使用pandas读取csv数据,然后将其转化为字典,再写入到数据库的时候,数据库总是报错类型错误,于是转为orm之前,统一转化一下类型
from sqlalchemy import DECIMAL, Index, String, Date, Integer, Text, CHAR, SmallInteger, Float, Time, case, and_, extract, TypeDecorator
from sqlalchemy.ext.hybrid import hybrid_property
from sqlalchemy.dialects.mysql import DATETIME
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column
from sqlalchemy import func, case, inspect
def trans_orm_type(v, k_orm_type, k):
if v is None or v == '':
return None
elif isinstance(k_orm_type, Date):
if isinstance(v, date):
return v
else:
if isinstance(v, datetime):
return v.date()
else:
if isinstance(v, int):
v = str(copy(v))
if len(v) == 6:
return parser.parse(v + '01').date()
else:
return parser.parse(v).date()
elif isinstance(k_orm_type, String):
if isinstance(v, str):
return v
else:
return str(v)
# SmallInteger 是 Integer的子类,所以判断逻辑需要放在Integer
elif isinstance(k_orm_type, SmallInteger):
if isinstance(v, int):
return v
else:
if v == '是':
return 1
elif v == '否':
return 0
else:
return int(v)
elif isinstance(k_orm_type, Integer):
if isinstance(v, int):
return v
else:
return int(v)
elif isinstance(k_orm_type, Float):
if isinstance(v, float):
return v
else:
return float(v)
else:
return v
class Base(DeclarativeBase):
def all_columns(self):
return [getattr(self, c.name, None) for c in self.__table__.columns]
def to_dict(self):
return {c.name: getattr(self, c.name, None) for c in self.__table__.columns}
@classmethod
def from_dict(cls, data):
data = {k: trans_orm_type(v, cls.__table__.columns[k].type, k) for k, v in data.items() if k in cls.__table__.columns}
return cls(**data)
标签:sqlite,sqlalchemy,return,else,orm,报错,isinstance,type
From: https://www.cnblogs.com/meizhengchao/p/18302118