選擇*在SQLAlchemy的

Question

我奮力運行SELECT *在SQL鍊金術查詢，我的表對象看起來像這樣，

from sqlalchemy import Column, Date, DateTime, Float, Numeric, Index, Integer, MetaData, SmallInteger, String, UnicodeText, Table, Time, text 
from sqlalchemy.types import _Binary 

metadata = MetaData() 

absence = Table(
    'absence', metadata, 
    Column('id', Integer, primary_key=True), 
    Column('workbase', String(255), index=True), 
    Column('name', String(255), nullable=False, index=True), 
    Column('submit_date', DateTime, index=True), 
    Column('created_1', DateTime, index=True), 
    Column('manager_1', String(255), index=True), 
    Column('first_day_1', Date, index=True), 
    Column('last_day_1', Date, index=True), 
    Column('sp_absence_id', Integer, index=True), 
    Column('created_2', DateTime, index=True), 
    Column('manager_2', String(255), index=True), 
    Column('first_day_2', Date, index=True), 
    Column('last_day_2', Date, index=True), 
    Column('sp_selfcert_id', Integer, index=True), 
    Column('sp_selfcert_file', String(255), index=True), 
    Column('interview_date', Date, index=True), 
    Column('created_3', DateTime, index=True), 
    Column('manager_3', String(255), index=True), 
    Column('first_day_3', Date, index=True), 
    Column('sp_return_id', Integer, index=True), 
    Column('sp_return_file', String(255)), 
    Column('rostered_days', Float(24), index=True), 
    Column('reason', String(255), nullable=False, index=True), 
    Column('rtw_trigger', Integer, index=True) 
) 

我曾嘗試選擇所有

absence = tables.absence 
#Attempt 1 
absence.all() 
returns AttributeError: 'Table' object has no attribute 'all' 
#Attempt 2 
absence.query.all() 
returns AttributeError: 'Table' object has no attribute 'query' 
#Attempt 3 
sel = select([absence.c.*]).select_from(absence) 
absences = FlaskApp.db_connect().execute(sel).fetchall() 
returns nothing, but kills my local server. 

無的無數種方法上面的工作，我做錯了我以前在其他語言中使用ORM，他們一直很容易，

Answer 1

我相信你錯過了數據庫創建階段描述他再次：http://docs.sqlalchemy.org/en/rel_1_1/core/metadata.html#creating-and-dropping-database-tables

嘗試（使用RAM源碼爲例）補充說：

engine = create_engine('sqlite:///:memory:') 
metadata.create_all(engine) 

然後你就可以按照這裏所描述的指令：http://docs.sqlalchemy.org/en/rel_1_1/core/tutorial.html#sql-expression-language-tutorial

順便說一句，在ORM機制（至極，不使用你的代碼示例），這裏有很好的描述：http://docs.sqlalchemy.org/en/rel_1_1/orm/tutorial.html

Answer 2

，我建議你用這個結構：

第1步：創建一個meta.py文件，它是連接到數據庫：

# a. Connect to the mydb database: 
from sqlalchemy import create_engine 
from settings import info # import info dictionary 

engine = create_engine('mysql://{user_name_db}:{password_db}@{host_name_db}:{port_db}/{db_name}?charset=utf8'.format(
    user_name_db=info['user_name_db'], 
    password_db=info['password_db'], 
    host_name_db=info['host_name_db'], 
    port_db=info['port_db'], 
    db_name=info['db_name'] 
), echo=False) 

# b. Call the session which bind the db engine to manipulate the database: 
from sqlalchemy.orm import sessionmaker 

Session = sessionmaker(bind=engine) 
session = Session() 

# c. Define the models class mapper: 
from sqlalchemy.ext.declarative import declarative_base 

Base = declarative_base() 

第2步：創建一個tables.py文件，它是你的數據庫表的模型：

from sqlalchemy import (Column, Integer, String, DateTime, Date, Float) 
from meta import * 


class Absence(Base): 
    __tablename__ = 'absence' 

    id = Column(Integer, primary_key=True) 
    workbase = Column(String(255), index=True) 
    name = Column(String(255), nullable=False, index=True) 
    submit_date = Column(DateTime, index=True) 
    created_1 = Column(DateTime, index=True) 
    manager_1 = Column(String(255), index=True) 
    first_day_1 = Column(Date, index=True) 
    last_day_1 = Column(Date, index=True) 
    sp_absence_id = Column(Integer, index=True) 
    created_2 = Column(DateTime, index=True) 
    manager_2 = Column(String(255), index=True) 
    first_day_2 = Column(Date, index=True) 
    last_day_2 = Column(Date, index=True) 
    sp_selfcert_id = Column(Integer, index=True) 
    sp_selfcert_file = Column(String(255), index=True) 
    interview_date = Column(Date, index=True) 
    created_3 = Column(DateTime, index=True) 
    manager_3 = Column(String(255), index=True) 
    first_day_3 = Column(Date, index=True) 
    sp_return_id = Column(Integer, index=True) 
    sp_return_file = Column(String(255)) 
    rostered_days = Column(Float(24), index=True) 
    reason = Column(String(255), nullable=False, index=True) 
    rtw_trigger = Column(Integer, index=True) 

    # initialize MyTable_model class 
    def __init__(self, id_=None, workbase=None, name=None, submit_date=None, created_1=None, manager_1=None, 
       first_day_1=None, last_day_1=None, sp_absence_id=None, created_2=None, manager_2=None, 
       first_day_2=None, last_day_2=None, sp_selfcert_id=None, sp_selfcert_file=None, interview_date=None, 
       created_3=None, manager_3=None, first_day_3=None, sp_return_id=None, sp_return_file=None, 
       rostered_days=None, reason=None, rtw_trigger=None): 
     self.id = id_ 
     self.workbase = workbase 
     self.name = name 
     self.submit_date = submit_date 
     self.created_1 = created_1 
     self.manager_1 = manager_1 
     self.first_day_1 = first_day_1 
     self.last_day_1 = last_day_1 
     self.sp_absence_id = sp_absence_id 
     self.created_2 = created_2 
     self.manager_2 = manager_2 
     self.first_day_2 = first_day_2 
     self.last_day_2 = last_day_2 
     self.sp_selfcert_id = sp_selfcert_id 
     self.sp_selfcert_file = sp_selfcert_file 
     self.interview_date = interview_date, 
     self.created_3 = created_3 
     self.manager_3 = manager_3 
     self.first_day_3 = first_day_3 
     self.sp_return_id = sp_return_id 
     self.sp_return_file = sp_return_file 
     self.rostered_days = rostered_days 
     self.reason = reason 
     self.rtw_trigger = rtw_trigger 

第3步：創建一個driver.py文件，它是您的疑問：

from tables import * 


class MyAbsence(Absence): 

    # - Add a new record 
    def create(self): 
     try: 
      new_record = Absence(workbase=self.workbase, name=self.name, submit_date=self.submit_date) # .... 
      session.add(new_record) 
      session.flush() 
      session.refresh(new_record) 
      id_record = new_record.id 
      session.commit() 
      return id_record 
     except Exception as e: 
      print("ERROR : MyAbsence.create : " + str(e)) 
      return 0 

    @staticmethod 
    def get_all(): 
     try: 
      return [{"id": dic.id, "workbase": dic.workbase, "name": dic.name, "submit_date": dic.submit_date} # ... 
        for dic in session.query(Absence).all()] 
     except Exception as e: 
      print("ERROR : MyAbsence.get_all : " + str(e)) 
      return [] 

第4步：使用，例如：

object_abs = MyAbsence() 
ls = object_abs.get_all() 
print(ls)