Python 开发环境的准备以及一些常用类库模块的安装

from fastapi import FastAPI

app = FastAPI()

@app.get("/")
async def read_root():
    return {"Hello": "World"}

if __name__ == "__main__":
    import uvicorn
    uvicorn.run(app, host="0.0.0.0", port=8000)

from fastapi import FastAPI

app = FastAPI()

@app.get("/")
async def read_root():
    return {"message": "Hello, World"}

@app.get("/items/{item_id}")
async def read_item(item_id: int, q: str = None):
    return {"item_id": item_id, "q": q}

if __name__ == "__main__":
    import uvicorn
    # log_level:'critical', 'error', 'warning', 'info', 'debug', 'trace'。默认值：'info'。
    uvicorn.run(
        app,
        host=settings.SERVER_IP,
        port=settings.SERVER_PORT,
        log_config="app/uvicorn_config.json",  # 日志配置
        # log_level="info",  # 日志级别
    )

pip install pymysql

import pymysql

connection = pymysql.connect(
    host='localhost',
    user='your_username',
    password='your_password',
    database='your_database'
)

try:
    with connection.cursor() as cursor:
        # 执行 SQL 查询
        cursor.execute("SELECT VERSION()")
        
        # 获取查询结果
        result = cursor.fetchone()
        print(f"MySQL version: {result}")
finally:
    connection.close()

sql = "select * from t_customer where name = '{0}' LIMIT 1 ".format(name)
print(sql)
cursor.execute(sql)
myresult = cursor.fetchone()  # fetchone() 获取一条记录

if myresult:
    print("该名称已存在，请更换名称.")
else:
    print("该名称可用.")

    # 插入记录语句
    sql = "INSERT INTO `t_customer` (`ID`, `Name`, `Age`, `Creator`, `CreateTime`) VALUES (%s, %s, %s, %s, %s);"
    val = (id, name, age, creator, createtime)

    cursor.execute(sql, val)
    db.commit()  # 数据表内容有更新，必须使用到该语句

    print(cursor.rowcount, " 行记录插入.")

sql = "update t_customer Set age = %s where name =%s  "
val = (26, name)
cursor.execute(sql, val)
db.commit()  # 数据表内容有更新，必须使用到该语句
print(cursor.rowcount, " 条记录被修改")

sql = "select * from t_customer where name = '{0}' LIMIT 1 ".format(name)
cursor.execute(sql)
myresult = cursor.fetchone()  # fetchone() 获取一条记录

if myresult:
    print("修改后的记录: ", myresult)

sql = "SELECT * FROM t_customer"
cursor.execute(sql)
print("t_customer 结果集: ")
for x in cursor:
    print(x)

sql = "delete from t_customer where name =%s  "
try:
    cursor.execute(sql, (name,))
    db.commit()  # 数据表内容有更新，必须使用到该语句
    print(cursor.rowcount, " 行记录删除.")
except:
    db.rollback()  # 发生错误时回滚
    print("删除记录失败!")


sql = "SELECT * FROM t_customer"
cursor.execute(sql)
myresult = cursor.fetchall()  # fetchall() 获取所有记录

for x in myresult:
    print(x)

# 关闭数据库连接
db.close()

pip install pymssql

import pymssql

# Connect to the database
conn = pymssql.connect(
    server="localhost",
    user="sa",
    password="123456",
    database="Winframework",
    tds_version="7.0",
)

# Create a cursor object
cursor = conn.cursor()

# Execute a query
cursor.execute("SELECT * FROM T_Customer")
# Fetch all the rows
rows = cursor.fetchall()

# Print the rows
for row in rows:
    print(row)


# Close the cursor and connection
cursor.close()
conn.close()

pip install sqlalchemy

# mysql 数据库引擎
engine = create_engine(
    "mysql+pymysql://root:123456@127.0.0.1:3306/WinFramework",
    pool_recycle=3600,
    # echo=True,
)

# Sqlite 数据库引擎
# engine = create_engine("sqlite:///testdir//test.db")

# PostgreSQL 数据库引擎
# engine = create_engine(
#     "postgresql+psycopg2://postgres:123456@localhost:5432/winframework",
#     # echo=True,
# )

# engine = create_engine(
#     "mssql+pymssql://sa:123456@localhost/WinFramework?tds_version=7.0",
#     # echo=True,
# )

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

# 创建一个基类，用于定义数据库表的结构
Base = declarative_base()

# 定义一个 Customer数据库表的模型
class Customer(Base):
    __tablename__ = "t_customer"

    id = Column(String, primary_key=True, comment="主键")
    name = Column(String, comment="姓名")
    age = Column(Integer, comment="年龄")
    creator = Column(String, comment="创建人")
    createtime = Column(DateTime, comment="创建时间")

# 创建一个会话
Session = sessionmaker(bind=engine)
session = Session()

id = str(guid.uuid4())
# create a new customer
customer = Customer(
    id=id,
    name="Alice",
    age=25,
    creator="admin",
    createtime=datetime.strptime("2021-01-01 12:00:00", "%Y-%m-%d %H:%M:%S"),
)

# add the customer to the session
session.add(customer)
# commit the changes to the database
session.commit()

# query the customer from the session
for item in session.scalars(select(Customer)):
    print(item.id, item.name, item.age, item.creator, item.createtime)


print("\r\nquery all customers")
customers = session.query(Customer).all()
for customer in customers:
    print(customer.name, customer.age)

print("\r\nquery all customers by condition:age > 20")
customers = session.query(Customer).filter(Customer.age > 20).limit(30).offset(1).all()
for customer in customers:
    print(customer.name, customer.age)

print("\r\nquery customer by id")
customer = session.query(Customer).filter(Customer.id == id).first()
if customer:
    print(customer.name, customer.age)

print("\r\n 复杂查询")
customers = (
    session.query(Customer)
    .filter(
        or_(
            and_(Customer.age > 20, Customer.age < 30),
            Customer.name.in_(["Alice", "伍华聪"]),
        )
    )
    .all()
)
for customer in customers:
    print(customer.name, customer.age)


print("\r\nselect customer by id")
stmt = select(Customer).where(Customer.id == id)
result = session.execute(stmt)
print(result)


stmt = select(Customer).where(Customer.name == "伍华聪")
result = session.execute(stmt).scalar()
if result:
    print("Customer exists in the database")
    print(result.id, result.name, result.age)
else:
    print("Customer does not exist in the database")


print("\r\nselect customer In")
# query the customer from the session
stmt = select(Customer).where(Customer.name.in_(["Alice", "伍华聪"]))
for item in session.scalars(stmt):
    print(item.id, item.name, item.age, item.creator, item.createtime)


print('\r\ndelete all customers by name = "Alice"')
# delete the customer from the database
delete_stmt = delete(Customer).where(Customer.name == "Alice")
result = session.execute(delete_stmt)
print(str(result.rowcount) + " rows deleted")

session.commit()

# close the session
session.close()

class Person(BaseModel):
    name: str
    age: int

    @field_validator("age")
    def age_must_be_positive(cls, v):
        if v < 0:
            raise ValueError("Age must be a positive number")
        return v

from pydantic_settings import BaseSettings, SettingsConfigDict

class Settings(BaseSettings):
    model_config = SettingsConfigDict(
        env_file=".env",  # 加载env文件
        extra="ignore",  # 加载env文件，如果没有在Settings中定义属性，也不抛出异常
        env_file_encoding="utf-8",
        env_prefix="",
        case_sensitive=False,
    )

    # Env Server
    SERVER_IP: str = "127.0.0.1"
    SERVER_PORT: int = 9000

    # Env Database
    DB_NAME: str = "winframework"
    DB_USER: str = "root"
    DB_PASSWORD: str = "123456"
    DB_HOST: str = "localhost"
    DB_PORT: int = 3306
    DB_URI: str = (
        f"mysql+pymysql://{DB_USER}:{DB_PASSWORD}@{DB_HOST}:{DB_PORT}/{DB_NAME}"
    )
   .............