python 数据库编程（mariadb）

from mysql.connector import connect
db_name = {
    "host": "localhost",
    "port": 3306,
    "user": "root",
    "password": "123456",
    "database": "test",
}

sql_string_dict = {
    "create": ‘create table genral_tab1 (card_id integer,name text,address text)‘,
    "insert":
    [‘insert into genral_tab1 (card_id,name,address) values (1,"LiMing","East Zone")‘,
     ‘insert into genral_tab1 (card_id,name,address) values (2,"WangMing","West Zone")‘,
     ‘insert into genral_tab1 (card_id,name,address) values (3,"ZhaoMing","South Zone")‘,
     ‘insert into genral_tab1 (card_id,name,address) values (4,"DingMing","North Zone")‘],
    "select": ‘select * from genral_tab1‘,
    "update": ‘update genral_tab1 set name="QianMing" where card_id=4‘,
    "delete": ‘delete from genral_tab1 where card_id=3‘,
}
con = connect(**db_name)     #建立连接,使用关键字参数,字典
cur = con.cursor()      #获取游标

print("create Table:\n")
cur.execute(sql_string_dict["create"])
print("create successfully!!")

print("insert Date:")
print(‘(1,"LiMing","East Zone")‘)
print(‘(2,"WangMing","West Zone")‘)
print(‘(3,"ZhaoMing","South Zone")‘)
print(‘(4,"DingMing","North Zone")‘)
for sql in sql_string_dict["insert"]:
    cur.execute(sql)
print()

print(‘All Records:‘)
cur.execute(sql_string_dict["select"])
for row in cur:
    print(row)
print()

cur.execute(sql_string_dict["update"])

print("After update")
cur.execute(sql_string_dict["select"])
for row in cur:
    print(row)
print()

cur.execute(sql_string_dict["delete"])
print("After delete")
cur.execute(sql_string_dict["select"])
for row in cur:
    print(row)
con.commit()    #提交数据，保存到数据库中
con.close()     #关闭数据库

from sqlalchemy import create_engine,String,Integer,Column
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker

db_name = ‘mysql+mysqlconnector://root:123456@localhost:3306/test‘  #‘数据库类型+数据库驱动名称://用户名:密码@机器地址:端口号/数据库名‘
Base = declarative_base()  #创建对象基类

#定义User对象
class User(Base):
    __tablename__ = ‘user‘       #表的名字
    id = Column(Integer,primary_key=True)      #表的结构
    name = Column(String(50))

engine = create_engine(db_name)    #初始化数据库连接
Base.metadata.create_all(engine)     #创建表

Session  = sessionmaker(bind=engine)    #创建DBSession类型
session = Session()    #实例化会话

#new_user = User(id=5,name=‘bob‘)  #创建User对象
#session.add(new_user)       #添加到session
#session.commit()    #提交保存数据库

user = session.query(User).filter(User.id==5).one() #创建Query查询，filter是where条件，one()返回一行
print(‘type:‘, type(user))
print(‘name:‘, user.name)

session.close()

‘‘‘
运行结果：
type: <class ‘__main__.User‘>
name: bob
‘‘‘

from sqlalchemy import create_engine,String,Integer,Column,ForeignKey
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker,relationship
#一对多关系
db_name = ‘mysql+mysqlconnector://root:123456@localhost:3306/test‘

Base = declarative_base()

class User(Base):
    __tablename__ = ‘user‘
    id = Column(Integer,primary_key=True)
    name = Column(String(50))
    addresses = relationship("Address",backref="itsuser")
class Address(Base):
    __tablename__ = ‘address‘
    id = Column(Integer,primary_key=True)
    uid = Column(Integer,ForeignKey(‘user.id‘))
    addrss = Column(String(100))

engine = create_engine(db_name)
Base.metadata.create_all(engine)

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

#在交互式环境运行
>>> u1 = session.query(User).all()
>>> aa = Address(id=1,addrss=‘my@qq.com‘)
>>> u1[0].addresses.append(aa)
>>> u1[0].addresses[0].addrss
‘my@qq.com‘