SQLite数据库操作:
import sqlite3
conn = sqlite3.connect('example.db')
c = conn.cursor()
# 创建表
c.execute('''
CREATE TABLE stocks
(date text, trans text, symbol text, qty real, price real)
''')
# 插入数据
c.execute("INSERT INTO stocks VALUES ('2023-10-05','BUY','RHAT',100,35.14)")
# 提交事务
conn.commit()
# 查询数据
c.execute("SELECT * FROM stocks WHERE symbol = 'RHAT'")
print(c.fetchall())
# 关闭连接
conn.close()
MySQL数据库操作:
import mysql.connector
conn = mysql.connector.connect(
host='localhost',
user='yourusername',
password='yourpassword',
database='yourdatabase'
)
cursor = conn.cursor()
# 创建表
cursor.execute('''CREATE TABLE stocks
(date VARCHAR(255), trans VARCHAR(255), symbol VARCHAR(255), qty DOUBLE, price DOUBLE)''')
# 插入数据
cursor.execute("INSERT INTO stocks (date, trans, symbol, qty, price) VALUES (%s, %s, %s, %s, %s)", ('2023-10-05', 'BUY', 'RHAT', 100, 35.14))
# 提交事务
conn.commit()
# 查询数据
cursor.execute("SELECT * FROM stocks WHERE symbol = 'RHAT'")
for row in cursor.fetchall():
print(row)
# 关闭连接
cursor.close()
conn.close()
PostgreSQL数据库操作:
import psycopg2
conn = psycopg2.connect(
host='localhost',
database='yourdatabase',
user='yourusername',
password='yourpassword'
)
cursor = conn.cursor()
# 创建表
cursor.execute('''CREATE TABLE stocks
(date VARCHAR(255), trans VARCHAR(255), symbol VARCHAR(255), qty DOUBLE PRECISION, price DOUBLE PRECISION)''')
# 插入数据
cursor.execute("INSERT INTO stocks (date, trans, symbol, qty, price) VALUES (%s, %s, %s, %s, %s)", ('2023-10-05', 'BUY', 'RHAT', 100, 35.14))
# 提交事务
conn.commit()
# 查询数据
cursor.execute("SELECT * FROM stocks WHERE symbol = 'RHAT'")
for row in cursor.fetchall():
print(row)
# 关闭连接
cursor.close()
conn.close()
SQLAlchemy ORM操作:
from sqlalchemy import create_engine, Column, Integer, String
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker
engine = create_engine('sqlite:///example.db')
Base = declarative_base()
class Person(Base):
__tablename__ = 'persons'
id = Column(Integer, primary_key=True)
name = Column(String)
age = Column(Integer)
Base.metadata.create_all(engine)
Session = sessionmaker(bind=engine)
session = Session()
# 插入数据
new_person = Person(name='Alice', age=30)
session.add(new_person)
session.commit()
# 查询数据
persons = session.query(Person).filter_by(name='Alice').all()
for person in persons:
print(person.id, person.name, person.age)
# 关闭会话
session.close()