sqlalchemy 행 객체를 python dict로 변환
열 이름과 값 쌍을 반복하는 간단한 방법이 있습니까?
내 sqlalchemy 버전은 0.5.6입니다
다음은 dict (row)를 사용하여 시도한 샘플 코드이지만 예외가 발생합니다. TypeError : 'User'객체를 반복 할 수 없습니다.
import sqlalchemy
from sqlalchemy import *
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker
print "sqlalchemy version:",sqlalchemy.__version__
engine = create_engine('sqlite:///:memory:', echo=False)
metadata = MetaData()
users_table = Table('users', metadata,
Column('id', Integer, primary_key=True),
Column('name', String),
)
metadata.create_all(engine)
class User(declarative_base()):
__tablename__ = 'users'
id = Column(Integer, primary_key=True)
name = Column(String)
def __init__(self, name):
self.name = name
Session = sessionmaker(bind=engine)
session = Session()
user1 = User("anurag")
session.add(user1)
session.commit()
# uncommenting next line throws exception 'TypeError: 'User' object is not iterable'
#print dict(user1)
# this one also throws 'TypeError: 'User' object is not iterable'
for u in session.query(User).all():
print dict(u)
내 시스템 출력에서이 코드를 실행합니다.
sqlalchemy version: 0.5.6
Traceback (most recent call last):
File "untitled-1.py", line 37, in <module>
print dict(u)
TypeError: 'User' object is not iterable
__dict__다음과 같이 SQLAlchemy 개체 의 내부 에 액세스 할 수 있습니다.
for u in session.query(User).all():
print u.__dict__
나는 좋은 대답을 얻지 못해서 이것을 사용합니다 :
def row2dict(row):
d = {}
for column in row.__table__.columns:
d[column.name] = str(getattr(row, column.name))
return d
편집 : 위의 기능이 너무 길고 일부 취향에 적합하지 않은 경우 여기에 하나의 라이너가 있습니다 (python 2.7+)
row2dict = lambda r: {c.name: str(getattr(r, c.name)) for c in r.__table__.columns}
SQLAlchemy v0.8 이상에서는 검사 시스템을 사용하십시오 .
from sqlalchemy import inspect
def object_as_dict(obj):
return {c.key: getattr(obj, c.key)
for c in inspect(obj).mapper.column_attrs}
user = session.query(User).first()
d = object_as_dict(user)
참고 .key열 이름, 예를 들어, 다음과 같은 경우와 다를 수 있습니다 속성의 이름입니다 :
class_ = Column('class', Text)
이 방법은 또한 작동합니다 column_property.
의견에서 @zzzeek에 따라 :
"row"가 ORM 매핑 인스턴스가 아닌 핵심 행 개체라고 가정하면 최신 버전의 SQLAlchemy에 대한 정답입니다.
for row in resultproxy:
row_as_dict = dict(row)
행에는 _asdict()dict를 제공 하는 함수 가 있습니다.
In [8]: r1 = db.session.query(Topic.name).first()
In [9]: r1
Out[9]: (u'blah')
In [10]: r1.name
Out[10]: u'blah'
In [11]: r1._asdict()
Out[11]: {'name': u'blah'}
@balki가 언급했듯이 :
이 _asdict()메소드는 KeyedTuple 로 리턴되므로 특정 필드를 조회하는 경우 사용할 수 있습니다 .
In [1]: foo = db.session.query(Topic.name).first()
In [2]: foo._asdict()
Out[2]: {'name': u'blah'}
반면 열을 지정하지 않으면 @charlax에서 제공하는 방법과 같이 제안 된 다른 방법 중 하나를 사용할 수 있습니다. 이 방법은 2.7 이상에서만 유효합니다.
In [1]: foo = db.session.query(Topic).first()
In [2]: {x.name: getattr(foo, x.name) for x in foo.__table__.columns}
Out[2]: {'name': u'blah'}
오래된 질문이지만 이것이 구글에서 "sqlalchemy row to dict"에 대한 첫 번째 결과이기 때문에 더 나은 답변을 얻을 가치가 있습니다.
SqlAlchemy가 반환하는 RowProxy 개체에는 items () 메서드가 있습니다. http://docs.sqlalchemy.org/en/latest/core/connections.html#sqlalchemy.engine.RowProxy.items
단순히 (키, 값) 튜플 목록을 반환합니다. 따라서 다음을 사용하여 행을 dict로 변환 할 수 있습니다.
파이썬 <= 2.6에서 :
rows = conn.execute(query)
list_of_dicts = [dict((key, value) for key, value in row.items()) for row in rows]
파이썬> = 2.7에서 :
rows = conn.execute(query)
list_of_dicts = [{key: value for (key, value) in row.items()} for row in rows]
from sqlalchemy.orm import class_mapper
def asdict(obj):
return dict((col.name, getattr(obj, col.name))
for col in class_mapper(obj.__class__).mapped_table.c)
다음 함수가 다음에 추가된다고 가정하면 class User모든 열의 모든 키-값 쌍을 반환합니다.
def columns_to_dict(self):
dict_ = {}
for key in self.__mapper__.c.keys():
dict_[key] = getattr(self, key)
return dict_
다른 답변과 달리 객체의 Column클래스 수준에있는 속성 인 객체의 속성 만 반환 됩니다. 따라서 SQLalchemy 또는 _sa_instance_state다른 속성이 없거나 개체에 추가 한 속성 이 포함됩니다. 참고
편집 : 이것은 상속 된 열에서도 작동한다는 것을 잊어 버렸습니다.
hybrid_propery 확장
hybrid_property속성 도 포함 하려면 다음이 작동합니다.
from sqlalchemy import inspect
from sqlalchemy.ext.hybrid import hybrid_property
def publics_to_dict(self) -> {}:
dict_ = {}
for key in self.__mapper__.c.keys():
if not key.startswith('_'):
dict_[key] = getattr(self, key)
for key, prop in inspect(self.__class__).all_orm_descriptors.items():
if isinstance(prop, hybrid_property):
dict_[key] = getattr(self, key)
return dict_
여기서는 _속성을 액세스 hybrid_property하거나 단순히 표시하고 싶지 않기 때문에 열 을 숨기고 싶다는 표시 로 시작한다고 가정 합니다. 참고
Tipp all_orm_descriptors 은 또한 hybrid_method 와 AssociationProxy 도 포함 시키려면 리턴 합니다.
다른 답변에 대한 비고
속성 을 기반으로 한 모든 답변 (예 : 1 , 2 )은 __dict__단순히 객체의 모든 속성을 반환합니다. 이것은 당신이 원하는 훨씬 더 많은 속성이 될 수 있습니다. 슬프게도 여기에는 _sa_instance_state이 객체에 정의한 다른 속성 이 포함됩니다 .
함수를 기반으로하는 모든 답변 (예 : 1 , 2 ) 은 질문에서 와 같이 작업하도록 정의한 클래스 가 아닌 반환 된 SQLalchemy 행 객체 dict()에서만 작동합니다 .session.execute()class User
해결의 답 에 근거 row.__table__.columns은 확실히 하지 작동합니다. row.__table__.columnsSQL 데이터베이스의 열 이름을 포함합니다. 이들은 python 객체의 속성 이름과 같을 수 있습니다 . 그렇지 않으면을 얻습니다 AttributeError. 답변 내용 (같은 1 , 2 페이지) 기반 class_mapper(obj.__class__).mapped_table.c이 동일하다.
@balki 답변에 이어 SQLAlchemy 0.8부터 KeyedTuple 객체에 사용 가능한 _asdict ()를 사용할 수 있습니다. 이것은 원래 질문에 대한 간단한 답변을 제공합니다. 예를 들어이 두 줄의 마지막 두 줄 (for 루프)을 변경하십시오.
for u in session.query(User).all():
print u._asdict()
위의 코드에서 u는 KeyedTuple 유형의 객체이기 때문에 작동합니다. .all ()은 KeyedTuple의 목록을 반환하기 때문입니다. 따라서 _asdict () 메소드가 있으며 u를 사전으로 멋지게 반환합니다.
@STB의 대답 : AFAIK, .all ()이 반환하는 anithong은 KeypedTuple의 목록입니다. 따라서 위의 내용은 Query 개체에 적용된 .all () 결과를 처리하는 한 열을 지정하거나 지정하지 않은 경우 작동합니다.
반복하는 표현식은 행이 아니라 모델 객체 목록으로 평가됩니다 . 따라서 다음은 올바른 사용법입니다.
for u in session.query(User).all():
print u.id, u.name
그것들을 dicts로 변환해야합니까? 물론 많은 방법이 있지만 SQLAlchemy의 ORM 부분은 필요하지 않습니다.
result = session.execute(User.__table__.select())
for row in result:
print dict(row)
업데이트 : sqlalchemy.orm.attributes모듈을 살펴보십시오 . 객체 상태와 함께 작동하는 함수 세트가 있으며 특히 유용합니다 instance_dict().
SQLAlchemy 행을 dict로 변환하는 방법을 찾고 있었기 때문에이 게시물을 찾았습니다. SqlSoup을 사용하고 있습니다 ...하지만 대답은 나 자신에 의해 작성되었으므로 누군가 내 2 센트를 도울 수 있다면 :
a = db.execute('select * from acquisizioni_motes')
b = a.fetchall()
c = b[0]
# and now, finally...
dict(zip(c.keys(), c.values()))
Alex Brasetvik의 답변을 참조하십시오. 한 줄의 코드를 사용하여 문제를 해결할 수 있습니다
row_as_dict = [dict(row) for row in resultproxy]
Alex Brasetvik의 답변의 의견 섹션에서 SQLAlchemy의 작성자 인 zzzeek는 이것이 문제의 "올바른 방법"이라고 말했습니다.
이런 식으로 시도해 볼 수 있습니다.
for u in session.query(User).all():
print(u._asdict())
쿼리 개체의 dictonary 개체를 반환하는 쿼리 개체에 기본 제공 메서드를 사용합니다.
참조 : https://docs.sqlalchemy.org/en/latest/orm/query.html
sqlalchemy 객체를 이와 같은 사전으로 변환하여 json / dictionary로 반환 할 수 있습니다.
도우미 기능 :
import json
from collections import OrderedDict
def asdict(self):
result = OrderedDict()
for key in self.__mapper__.c.keys():
if getattr(self, key) is not None:
result[key] = str(getattr(self, key))
else:
result[key] = getattr(self, key)
return result
def to_array(all_vendors):
v = [ ven.asdict() for ven in all_vendors ]
return json.dumps(v)
운전사 기능 :
def all_products():
all_products = Products.query.all()
return to_array(all_products)
두 가지 방법:
1.
for row in session.execute(session.query(User).statement):
print(dict(row))
2.
selected_columns = User.__table__.columns
rows = session.query(User).with_entities(*selected_columns).all()
for row in rows :
print(row._asdict())
문서는 매우 간단한 솔루션을 제공합니다. ResultRow._asdict()
def to_array(rows):
return [r._asdict() for r in rows]
def query():
data = session.query(Table).all()
return to_array(data)
다음은 Elixir가 수행하는 방법입니다. 이 솔루션의 가치는 관계의 사전 표현을 재귀 적으로 포함 할 수 있다는 것입니다.
def to_dict(self, deep={}, exclude=[]):
"""Generate a JSON-style nested dict/list structure from an object."""
col_prop_names = [p.key for p in self.mapper.iterate_properties \
if isinstance(p, ColumnProperty)]
data = dict([(name, getattr(self, name))
for name in col_prop_names if name not in exclude])
for rname, rdeep in deep.iteritems():
dbdata = getattr(self, rname)
#FIXME: use attribute names (ie coltoprop) instead of column names
fks = self.mapper.get_property(rname).remote_side
exclude = [c.name for c in fks]
if dbdata is None:
data[rname] = None
elif isinstance(dbdata, list):
data[rname] = [o.to_dict(rdeep, exclude) for o in dbdata]
else:
data[rname] = dbdata.to_dict(rdeep, exclude)
return data
이 코드를 사용하면 쿼리 "filter"또는 "join"에이 작업을 추가 할 수도 있습니다.
query = session.query(User)
def query_to_dict(query):
def _create_dict(r):
return {c.get('name'): getattr(r, c.get('name')) for c in query.column_descriptions}
return [_create_dict(r) for r in query]
class User(object):
def to_dict(self):
return dict([(k, getattr(self, k)) for k in self.__dict__.keys() if not k.startswith("_")])
작동합니다.
Marco Mariani의 답변에 변형이있어 장식 자로 표현되었습니다. 가장 큰 차이점은 엔티티 목록 을 처리 하고 다른 유형의 반환 값을 안전하게 무시한다는 것입니다 (모의를 사용하여 테스트를 작성할 때 매우 유용합니다).
@decorator
def to_dict(f, *args, **kwargs):
result = f(*args, **kwargs)
if is_iterable(result) and not is_dict(result):
return map(asdict, result)
return asdict(result)
def asdict(obj):
return dict((col.name, getattr(obj, col.name))
for col in class_mapper(obj.__class__).mapped_table.c)
def is_dict(obj):
return isinstance(obj, dict)
def is_iterable(obj):
return True if getattr(obj, '__iter__', False) else False
@Anurag Uniyal의 답변을 완료하려면 다음과 같이 관계를 재귀 적으로 따르는 방법이 있습니다.
from sqlalchemy.inspection import inspect
def to_dict(obj, with_relationships=True):
d = {}
for column in obj.__table__.columns:
if with_relationships and len(column.foreign_keys) > 0:
# Skip foreign keys
continue
d[column.name] = getattr(obj, column.name)
if with_relationships:
for relationship in inspect(type(obj)).relationships:
val = getattr(obj, relationship.key)
d[relationship.key] = to_dict(val) if val else None
return d
class User(Base):
__tablename__ = 'users'
id = Column(Integer, primary_key=True)
first_name = Column(TEXT)
address_id = Column(Integer, ForeignKey('addresses.id')
address = relationship('Address')
class Address(Base):
__tablename__ = 'addresses'
id = Column(Integer, primary_key=True)
city = Column(TEXT)
user = User(first_name='Nathan', address=Address(city='Lyon'))
# Add and commit user to session to create ids
to_dict(user)
# {'id': 1, 'first_name': 'Nathan', 'address': {'city': 'Lyon'}}
to_dict(user, with_relationship=False)
# {'id': 1, 'first_name': 'Nathan', 'address_id': 1}
나는 새로 개발 된 Python 프로그래머이며 Joined 테이블을 사용하여 JSON을 얻는 데 문제가 발생했습니다. 여기에있는 답변의 정보를 사용하여 별칭을 사용하지 않거나 필드 충돌을 피하면서 테이블 이름이 포함되는 합리적인 결과를 JSON으로 반환하는 함수를 작성했습니다.
세션 쿼리 결과를 간단히 전달하십시오.
test = Session (). query (VMInfo, Customer) .join (Customer) .order_by (VMInfo.vm_name) .limit (50) .offset (10)
json = sqlAl2json (테스트)
def sqlAl2json(self, result):
arr = []
for rs in result.all():
proc = []
try:
iterator = iter(rs)
except TypeError:
proc.append(rs)
else:
for t in rs:
proc.append(t)
dict = {}
for p in proc:
tname = type(p).__name__
for d in dir(p):
if d.startswith('_') | d.startswith('metadata'):
pass
else:
key = '%s_%s' %(tname, d)
dict[key] = getattr(p, d)
arr.append(dict)
return json.dumps(arr)
모델 테이블 열이 mysql 열이 아닌 경우.
같은 :
class People:
id: int = Column(name='id', type_=Integer, primary_key=True)
createdTime: datetime = Column(name='create_time', type_=TIMESTAMP,
nullable=False,
server_default=text("CURRENT_TIMESTAMP"),
default=func.now())
modifiedTime: datetime = Column(name='modify_time', type_=TIMESTAMP,
server_default=text("CURRENT_TIMESTAMP"),
default=func.now())
사용해야합니다 :
from sqlalchemy.orm import class_mapper
def asDict(self):
return {x.key: getattr(self, x.key, None) for x in
class_mapper(Application).iterate_properties}
이 방법을 사용하면 modify_time 및 create_time을 모두 얻을 수 있습니다.
{'id': 1, 'create_time': None, 'modify_time': None}
def to_dict(self):
return {c.name: getattr(self, c.name, None)
for c in self.__table__.columns}
mysql의 클래스 속성 이름이 열 저장소와 같지 않기 때문에
이 클래스의 내용을 .KeyedTuple사전으로 반환하십시오.
In [46]: result = aggregate_events[0]
In [47]: type(result)
Out[47]: sqlalchemy.util._collections.result
In [48]: def to_dict(query_result=None):
...: cover_dict = {key: getattr(query_result, key) for key in query_result.keys()}
...: return cover_dict
...:
...:
In [49]: to_dict(result)
Out[49]:
{'calculate_avg': None,
'calculate_max': None,
'calculate_min': None,
'calculate_sum': None,
'dataPointIntID': 6,
'data_avg': 10.0,
'data_max': 10.0,
'data_min': 10.0,
'data_sum': 60.0,
'deviceID': u'asas',
'productID': u'U7qUDa',
'tenantID': u'CvdQcYzUM'}
모두와 자신을 위해 여기에 내가 사용하는 방법이 있습니다.
def run_sql(conn_String):
output_connection = engine.create_engine(conn_string, poolclass=NullPool).connect()
rows = output_connection.execute('select * from db1.t1').fetchall()
return [dict(row) for row in rows]
def to_dict(row):
return {column.name: getattr(row, row.__mapper__.get_property_by_column(column).key) for column in row.__table__.columns}
for u in session.query(User).all():
print(to_dict(u))
이 기능이 도움이 될 수 있습니다. 속성 이름이 열 이름과 다를 때 문제를 해결하는 더 나은 솔루션을 찾을 수 없습니다.
프로젝트의 어느 곳에서나 필요합니다. @anurag가 잘 작동한다고 대답했습니다. 이 시점까지 나는 그것을 사용하고 있었지만 모든 코드를 망칠 것이고 엔티티 변경으로 작동하지 않을 것입니다.
오히려 이것을 시도하고 SQLAlchemy에서 기본 쿼리 클래스를 상속하십시오.
from flask_sqlalchemy import SQLAlchemy, BaseQuery
class Query(BaseQuery):
def as_dict(self):
context = self._compile_context()
context.statement.use_labels = False
columns = [column.name for column in context.statement.columns]
return list(map(lambda row: dict(zip(columns, row)), self.all()))
db = SQLAlchemy(query_class=Query)
그 후에 객체를 정의 할 때마다 "as_dict"메소드가 있습니다.
대부분의 시나리오에서 열 이름이 적합합니다. 그러나 다음과 같이 코드를 작성할 수 있습니다.
class UserModel(BaseModel):
user_id = Column("user_id", INT, primary_key=True)
email = Column("user_email", STRING)
column.name "user_email"이고 필드 이름이 "email"인 경우 column.name이 이전과 같이 제대로 작동하지 않습니다.
나는 이것에 대해 많은 경험이 없지만 다음은 내가하고있는 일에 효과가있는 것 같습니다.
dict(row)
이것은 너무 간단 해 보입니다 (여기의 다른 답변과 비교). 내가 무엇을 놓치고 있습니까?
참고 URL : https://stackoverflow.com/questions/1958219/convert-sqlalchemy-row-object-to-python-dict
'IT story' 카테고리의 다른 글
| 서로 옆에 CSS 두 div (0) | 2020.04.26 |
|---|---|
| HttpClient의 HttpRequestMessage에서 쿠키를 설정하는 방법 (0) | 2020.04.26 |
| ADB 장치를 찾을 수 없습니다 (0) | 2020.04.26 |
| Xcode 오류 : SDK 'iOS 10.0'의 제품 유형 'Application'에 코드 서명이 필요합니다. (0) | 2020.04.26 |
| OR (||)과 같이 스위치에서 여러 사례를 테스트 (0) | 2020.04.26 |