sqlalchemy 的显式 ORM 访问方式
对于sqlalchemy,可以利用一种显式的ORM方式进行访问,这种方式无需依赖声明层,而是显式地进行操作。还有一种访问方式为声明层 ORM 访问方式。
主要的建立步骤包括:
1. 设置DSN连接网址;
2. 创建引擎,利用引擎对DSN进行连接;
3. 创建元数据实例,并绑定到引擎;
4. 创建表类,并与元数据进行绑定;
5. 进行数据库操作。
基本连接关系图:
完整代码
![](https://images.cnblogs.com/OutliningIndicators/ContractedBlock.gif)
1 from sqlalchemy import create_engine, MetaData 2 from sqlalchemy.ext.declarative import declarative_base 3 from sqlalchemy.sql.schema import Table, Column 4 from sqlalchemy.sql.sqltypes import Integer, VARCHAR 5 from sqlalchemy.orm import sessionmaker 6 7 # No need to create declarative_base 8 # Base = declarative_base() 9 10 # Set DSN 11 dsn = 'mysql+mysqldb://root:root@localhost/test_db' 12 13 # Create engine, get connection 14 eng = create_engine(dsn) 15 cnx = eng.connect() 16 17 # No need to create session 18 # Session = sessionmaker(bind=eng) 19 # session = Session() 20 21 # Generate metadata instance 22 metadata = MetaData() 23 # Bind engine to metadata 24 metadata.bind = eng 25 26 # Build up Table 27 test_tb = Table('test_tb', metadata, 28 Column('id', Integer), 29 Column('name', VARCHAR(8)) 30 ) 31 #print(type(test_tb.insert())) 32 # Insert value, .rowcount will return the row number affected 33 print(test_tb.insert().execute(id=6, name='Momo').rowcount) 34 # Delete value, note: '.c' attr should be added 35 print(test_tb.delete(test_tb.c.id==6).execute().rowcount) 36 37 print(test_tb.insert().execute([{'id': 7, 'name': 'Momo'}, {'id': 8, 'name': 'Kitkat'}]).rowcount) 38 # Update value, change value where id is 7 to 9 39 print(test_tb.update(test_tb.c.id==7).execute(id=9).rowcount) 40 41 print(test_tb.delete(test_tb.c.id==8).execute().rowcount) 42 print(test_tb.delete(test_tb.c.id==9).execute().rowcount) 43 44 cnx.close()
分段分析
首先导入所需的模块,由于不需要声明层,此处进行注释
from sqlalchemy import create_engine, MetaData from sqlalchemy.ext.declarative import declarative_base from sqlalchemy.sql.schema import Table, Column from sqlalchemy.sql.sqltypes import Integer, VARCHAR from sqlalchemy.orm import sessionmaker # No need to create declarative_base # Base = declarative_base()
设置DSN参数,利用该参数进行引擎实例创建并连接,再创建元数据,与引擎进行绑定,最后建立表格类,与元数据进行绑定。
# Set DSN dsn = 'mysql+mysqldb://root:root@localhost/test_db' # Create engine, get connection eng = create_engine(dsn) cnx = eng.connect() # No need to create session # Session = sessionmaker(bind=eng) # session = Session() # Generate metadata instance metadata = MetaData() # Bind engine to metadata metadata.bind = eng # Build up Table test_tb = Table('test_tb', metadata, Column('id', Integer), Column('name', VARCHAR(8)) )
进行基本操作测试,注意此处删除时的条件语句在表格名和列名中间需要加入c表示列属性,最后利用连接实例关闭连接。
#print(type(test_tb.insert())) # Insert value, .rowcount will return the row number affected print(test_tb.insert().execute(id=6, name='Momo').rowcount) # Delete value, note: '.c' attr should be added print(test_tb.delete(test_tb.c.id==6).execute().rowcount) print(test_tb.insert().execute([{'id': 7, 'name': 'Momo'}, {'id': 8, 'name': 'Kitkat'}]).rowcount) # Update value, change value where id is 7 to 9 print(test_tb.update(test_tb.c.id==7).execute(id=9).rowcount) print(test_tb.delete(test_tb.c.id==8).execute().rowcount) print(test_tb.delete(test_tb.c.id==9).execute().rowcount) cnx.close()
相关阅读
1. 声明层 ORM 访问方式