import sqlite3 # 導(dǎo)?SQLite模塊

# 連接到SQLite數(shù)據(jù)庫(kù)
conn = sqlite3.connect('students.db')

# 創(chuàng)建?個(gè)游標(biāo)對(duì)象
cur = conn.cursor()

cur.execute(
    '''
    create table if not exists students(
        id integer primary key autoincrement,
        title text not null,
        author text not null,
        year integer
    )
    '''
)

cur.execute("insert into book(title,author,year) values('Python編程','張三',2022)")

cur.execute("delete from book where title = '算法導(dǎo)論'")

cur.execute("update book set year = 2023 where title = 'Python編程'")

cur.execute("select title,author from book")
cur.execute("select * from book")

cur.execute("select * from book")
data = cur.fetchall()
print(data)

cur.execute(f"select * from {self.excel} where name = ?",(name,))

import sqlite3 # 導(dǎo)?SQLite模塊
# 創(chuàng)建數(shù)據(jù)庫(kù)連接函數(shù)
def create_connection(db_file):
 """
 創(chuàng)建與SQLite數(shù)據(jù)庫(kù)的連接
 :param db_file: 數(shù)據(jù)庫(kù)?件路徑
 :return: 數(shù)據(jù)庫(kù)連接對(duì)象
 """
 conn = None # 定義?個(gè)連接對(duì)象
 try:
 conn = sqlite3.connect(db_file) # 嘗試連接到SQLite數(shù)據(jù)庫(kù)
 print("已成功創(chuàng)建連接")
 return conn # 如果成功，返回連接對(duì)象
 except sqlite3.Error as e: # 如果連接失敗，打印錯(cuò)誤信息
 print(f"錯(cuò)誤 '{e}' ")
# 創(chuàng)建表函數(shù)
def create_table(conn):
 """
 創(chuàng)建?戶表
 :param conn: 數(shù)據(jù)庫(kù)連接對(duì)象
 """
 try:
 cursor = conn.cursor() # 獲取游標(biāo)對(duì)象
 cursor.execute('''
 CREATE TABLE IF NOT EXISTS users(
 id INTEGER PRIMARY KEY, 
 name TEXT UNIQUE, 
 age INTEGER)
 ''')
 except sqlite3.Error as e:
 print(e) # 如果創(chuàng)建表失敗，打印錯(cuò)誤信息
# 插?數(shù)據(jù)函數(shù)
def insert_user(conn, data):
 """
 插??戶數(shù)據(jù)
 :param conn: 數(shù)據(jù)庫(kù)連接對(duì)象
 :param data: ?戶數(shù)據(jù)元組，包含name和age
 """
 try:
Python10
 cursor = conn.cursor() # 獲取游標(biāo)對(duì)象
 # 檢查?戶是否已存在
 cursor.execute("SELECT id FROM users WHERE name = ?", (data[0],))
 user_id = cursor.fetchone() # 獲取查詢結(jié)果
 if user_id is None: # 如果?戶不存在，插?新?戶
 cursor.execute("INSERT INTO users (name, age) VALUES (?, ?)",
 data)
 conn.commit() # 提交事務(wù)
 else:
 print(f"?戶： {data[0]} 的id： {user_id[0]}已存在")
 except sqlite3.Error as e:
 print(e) # 如果插?失敗，打印錯(cuò)誤信息
# 更新數(shù)據(jù)函數(shù)
def update_user(conn, data):
 """
 更新?戶數(shù)據(jù)
 :param conn: 數(shù)據(jù)庫(kù)連接對(duì)象
 :param data: ?戶數(shù)據(jù)元組，包含age和name
 """
 try:
 cursor = conn.cursor() # 獲取游標(biāo)對(duì)象
 cursor.execute("UPDATE users SET age = ? WHERE name = ?", data)
 conn.commit() # 提交事務(wù)
 except sqlite3.Error as e:
 print(e) # 如果更新失敗，打印錯(cuò)誤信息
# 刪除數(shù)據(jù)函數(shù)
def delete_user(conn, name):
 """
 刪除?戶數(shù)據(jù)
 :param conn: 數(shù)據(jù)庫(kù)連接對(duì)象
 :param name: ?戶名稱
 """
 try:
 cursor = conn.cursor() # 獲取游標(biāo)對(duì)象
 cursor.execute("DELETE FROM users WHERE name = ?", (name,))
 conn.commit() # 提交事務(wù)
 except sqlite3.Error as e:
 print(e) # 如果刪除失敗，打印錯(cuò)誤信息
# 查詢所有數(shù)據(jù)函數(shù)
def select_all_users(conn):
 """
 查詢所有?戶數(shù)據(jù)
 :param conn: 數(shù)據(jù)庫(kù)連接對(duì)象
 """
 try:
 cursor = conn.cursor() # 獲取游標(biāo)對(duì)象
 cursor.execute("SELECT * FROM users") # 執(zhí)?查詢所有?戶的SQL語句
 rows = cursor.fetchall() # 獲取所有查詢結(jié)果
 for row in rows: # 遍歷所有結(jié)果并打印
 print(row)
 except sqlite3.Error as e:
 print(e) # 如果查詢失敗，打印錯(cuò)誤信息
# 主函數(shù)
def main():
 database = "students2.db" # 定義數(shù)據(jù)庫(kù)?件名稱
 conn = create_connection(database) # 創(chuàng)建數(shù)據(jù)庫(kù)連接
 if conn is not None: # 如果連接成功
 create_table(conn) # 創(chuàng)建表
 try:
 insert_user(conn, ('Alice', 30))
 insert_user(conn, ('Bob', 25))
 # 更新數(shù)據(jù)
 update_user(conn, (35, 'Alice'))
 # 查詢并打印所有?戶
 print("查詢到的所有?戶:")
 select_all_users(conn)
 # 刪除?戶
 delete_user(conn, 'Bob')
 # 再次查詢并打印所有?戶
 print("\n刪除后的?戶數(shù)據(jù):")
 select_all_users(conn)
 except sqlite3.Error as e:
 print(f"出現(xiàn)錯(cuò)誤: {e}")
 finally:
 conn.close() # 確保關(guān)閉數(shù)據(jù)庫(kù)連接
 else:
 print("Error! ?法創(chuàng)建數(shù)據(jù)庫(kù)連接。")
# 程序??
if __name__ == '__main__':
 main()

import sqlite3
class DatabaseManager:
 """
 數(shù)據(jù)庫(kù)管理器類，?于處理與SQLite數(shù)據(jù)庫(kù)的交互
 """
 def __init__(self, db_file):
 """
 初始化數(shù)據(jù)庫(kù)管理器
 :param db_file: 數(shù)據(jù)庫(kù)?件路徑
 """
 self.db_file = db_file
 self.conn = None
 def create_connection(self):
 """
 創(chuàng)建與SQLite數(shù)據(jù)庫(kù)的連接
 """
 try:
 self.conn = sqlite3.connect(self.db_file)
 print("已成功創(chuàng)建連接")
 except sqlite3.Error as e:
 print(f"錯(cuò)誤 '{e}'")
 def create_table(self):
 """
 創(chuàng)建?戶信息表，包括id、name和age字段
 """
 if self.conn:
 try:
 cursor = self.conn.cursor()
 cursor.execute('''
 CREATE TABLE IF NOT EXISTS users(
 id INTEGER PRIMARY KEY AUTOINCREMENT,
 name TEXT UNIQUE,
 age INTEGER)
 ''')
 except sqlite3.Error as e:
 print(e)
 def insert_user(self, name, age):
 """
 向?戶表中插?新的?戶記錄
:param name: ?戶名
 :param age: ?戶年齡
 """
 if self.conn:
 try:
 cursor = self.conn.cursor()
 cursor.execute("SELECT id FROM users WHERE name = ?", (na
me,))
 if cursor.fetchone() is None:
 cursor.execute("INSERT INTO users (name, age) VALUES 
(?, ?)", (name, age))
 self.conn.commit()
 else:
 print(f"?戶：{name} 已存在")
 except sqlite3.Error as e:
 print(e)
 def update_user(self, name, age):
 """
 更新?戶表中?戶的年齡信息
 :param name: ?戶名
 :param age: 新的?戶年齡
 """
 if self.conn:
 try:
 cursor = self.conn.cursor()
 cursor.execute("UPDATE users SET age = ? WHERE name = ?",
 (age, name))
 self.conn.commit()
 except sqlite3.Error as e:
 print(e)
 def delete_user(self, name):
 """
 從?戶表中刪除指定的?戶
 :param name: 要?jiǎng)h除的?戶名
 """
 if self.conn:
 try:
 cursor = self.conn.cursor()
 cursor.execute("DELETE FROM users WHERE name = ?", (name,
))
 self.conn.commit()
 except sqlite3.Error as e:
 print(e)
 def select_all_users(self):
 """
 查詢并打印?戶表中所有?戶的信息
 """
 if self.conn:
 try:
 cursor = self.conn.cursor()
 cursor.execute("SELECT * FROM users")
 rows = cursor.fetchall()
 for row in rows:
 print(row)
 except sqlite3.Error as e:
 print(e)
 def close_connection(self):
 """
 關(guān)閉與數(shù)據(jù)庫(kù)的連接
 """
 if self.conn:
 self.conn.close()
 print("數(shù)據(jù)庫(kù)連接已關(guān)閉")
def main():
 """
 主函數(shù)，?于演示數(shù)據(jù)庫(kù)操作
 """
 database = "students3.db" # 定義數(shù)據(jù)庫(kù)?件名稱
 db_manager = DatabaseManager(database) # 創(chuàng)建數(shù)據(jù)庫(kù)管理器實(shí)例
 db_manager.create_connection() # 創(chuàng)建數(shù)據(jù)庫(kù)連接
 db_manager.create_table() # 創(chuàng)建?戶表
 try:
 db_manager.insert_user('Alice', 30) # 插??戶Alice
 db_manager.insert_user('Bob', 25) # 插??戶Bob
 db_manager.update_user('Alice', 35) # 更新?戶Alice的年齡
 print("查詢到的所有?戶:")
 db_manager.select_all_users() # 查詢并打印所有?戶
 db_manager.delete_user('Bob') # 刪除?戶Bob
 print("\n刪除后的?戶數(shù)據(jù):")
 db_manager.select_all_users() # 再次查詢并打印所有?戶
 except sqlite3.Error as e:
 print(f"出現(xiàn)錯(cuò)誤: {e}")
 finally:
 db_manager.close_connection() # 關(guān)閉數(shù)據(jù)庫(kù)連接
if __name__ == '__main__':
 main()

# 1.導(dǎo)入模塊
from fastapi import FastAPI, HTTPException
from pydantic import BaseModel
from sqlalchemy import create_engine, Column,Integer,String
from sqlalchemy.orm import sessionmaker, declarative_base

# 2.初始化FastAPI對(duì)象
app = FastAPI()

# 3.數(shù)據(jù)庫(kù)配置
# 定義了 SQLite 數(shù)據(jù)庫(kù)的連接路徑
database_url = "sqlite:///./text2.db"
# 創(chuàng)建數(shù)據(jù)庫(kù)連接
engine = create_engine(database_url, connect_args={"check_same_thread": False})
# 創(chuàng)建會(huì)話
# bind=engine ：指定要操作的數(shù)據(jù)庫(kù)連接
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)

# 4.創(chuàng)建數(shù)據(jù)庫(kù)模型
# 定義數(shù)據(jù)庫(kù)模型
Base = declarative_base()
class Login(Base):
    # 表名
    __tablename__ = "login"
    id = Column(Integer, primary_key=True, index=True)
    username = Column(String, unique=True, index=True)
    password = Column(String)

class User(Base):
    # 表名
    __tablename__ = "user"
    id = Column(Integer, primary_key=True, index=True)
    name = Column(String)
    age = Column(Integer)
    address = Column(String)
    numberphone = Column(String)

# 5.創(chuàng)建數(shù)據(jù)庫(kù)表
# 如果bind=engin的數(shù)據(jù)庫(kù)連接對(duì)應(yīng)的數(shù)據(jù)庫(kù)表不存在，則創(chuàng)建
Base.metadata.create_all(bind=engine)

# 6.創(chuàng)建用戶模型，用于接收請(qǐng)求的用戶數(shù)據(jù)
# BaseModel：基礎(chǔ)類
class LoginCreate(BaseModel):
    username: str
    password: str

class UserCreate(BaseModel):
    name: str
    age: int | None = None
    address: str | None = None
    numberphone: str | None = None

class UserSelect(BaseModel):
    flag: int
    name: str | None = None


# 使用下面代碼后，鼠標(biāo)右鍵直接運(yùn)行啟動(dòng)，不需要使用終端
if __name__ == "__main__":
    # 導(dǎo)?uvicorn模塊，?于啟動(dòng)FastAPI服務(wù)
    import uvicorn
    # 啟動(dòng)FastAPI服務(wù)，監(jiān)聽所有IP地址，端?號(hào)為8000
    # host="0.0.0.0" 表示監(jiān)聽所有IP地址
    # host="127.0.0.1" 表示監(jiān)聽本地IP地址
    uvicorn.run("main:app", host="127.0.0.1", port=9991)


@app.get("/")
async def root():
    # json類型
    return {"message": "Hello World"}

# ② 提供用戶登錄接口，當(dāng)用戶名和密碼正確時(shí)，提示：歡迎xxx，否則提示：用戶名或密碼不正確！
@app.get("/login")
def login(login: LoginCreate):
    try:
        session = SessionLocal()
        user_db = session.query(Login).filter(Login.username == login.username,
                                             Login.password == login.password).first()
        if user_db:
            return f"{login.username}登錄成功！"
        else:
            return "用戶名或密碼錯(cuò)誤"
    except Exception as e:
        return f"登錄失敗: {e}"



# ① 提供用戶注冊(cè)接口，當(dāng)用戶名和密碼不為空（注意：不能有相同的用戶名）
@app.post("/register")
def register(login: LoginCreate):
    try:
        session = SessionLocal()
        user_db = session.query(Login).filter(Login.username == login.username).first()
        if user_db:
            return "用戶已存在"
        else:
            session.add(Login(username=login.username, password=login.password))
            session.commit()
            return "注冊(cè)成功"
    except Exception as e:
        return f"注冊(cè)失敗: {e}"

@app.post("/user_select")
def user_select(user: UserSelect):
    session = SessionLocal()
    if user.flag == 1:
        # 查詢所有用戶信息
        return session.query(User).all()
    elif user.flag == 2:
        # 查詢單個(gè)用戶信息
        return session.query(User).filter(User.name == user.name).first()

@app.post("/user_update")
def user_update(user: UserCreate):
    session = SessionLocal()
    flag = session.query(User).filter(User.name == user.name).first()
    if flag:
        # 直接修改對(duì)象屬性
        flag.name = user.name
        flag.age = user.age
        flag.address = user.address
        flag.numberphone = user.numberphone
        session.commit()
        return "修改成功"
    else:
        return "用戶不存在"

@app.post("/user_delete")
def user_delete(user: UserCreate):
    session = SessionLocal()
    flag = session.query(User).filter(User.name == user.name).first()
    if flag:
        session.delete(flag)
        session.commit()
        return "刪除成功"
    else:
        return "用戶不存在"

@app.post("/user_insert")
def user_insert(user: UserCreate):
    session = SessionLocal()
    flag = session.query(User).filter(User.name == user.name).first()
    if flag:
        return "用戶已存在"
    else:
        session.add(User(name=user.name, age=user.age, address=user.address, numberphone=user.numberphone))
        session.commit()
        return "新增成功"

import sqlite3

import requests
import json

def user_table(n):
    if n == 1:
        # 查詢
        flag = input("輸入：1、查詢所有用戶信息頁面\t\t2、查詢單個(gè)用戶信息頁面")
        if flag == '1':
            k = {"flag": 1}
            res_get = requests.post("http://127.0.0.1:9991/user_select", json=k)
            return res_get.text
        elif flag == '2':
            name = input("請(qǐng)輸入用戶名：")
            k = {"flag": 2, "name": name}
            res_get = requests.post("http://127.0.0.1:9991/user_select", json=k)
            return res_get.text
        else:
            return
    elif n == 2:
        # 修改
        name = input("請(qǐng)輸入用戶名：")
        age = input("請(qǐng)輸入年齡：")
        address = input("請(qǐng)輸入地址：")
        numberphone = input("請(qǐng)輸入手機(jī)號(hào)：")
        params = {
            "name": name,
            "age": age,
            "address": address,
            "numberphone": numberphone
        }
        res_post = requests.post("http://127.0.0.1:9991/user_update", json=params)
        return res_post.text
    elif n == 3:
        # 刪除
        name = input("請(qǐng)輸入用戶名：")
        res_post = requests.post("http://127.0.0.1:9991/user_delete", json={"name": name})
        return res_post.text
    elif n == 4:
        # 新增
        name = input("請(qǐng)輸入用戶名：")
        age = input("請(qǐng)輸入年齡：")
        address = input("請(qǐng)輸入地址：")
        numberphone = input("請(qǐng)輸入手機(jī)號(hào)：")
        params = {
            "name": name,
            "age": age,
            "address": address,
            "numberphone": numberphone
        }
        res_post = requests.post("http://127.0.0.1:9991/user_insert", json=params)
        return res_post.text



#get
while True:
    flag = input("輸入：1、登錄頁面\t\t2、注冊(cè)頁面\t\t3、退出系統(tǒng)")
    if flag == '1':
        print("--------登錄頁面----------")
        username = input("請(qǐng)輸入用戶名：")
        password = input("請(qǐng)輸入密碼：")
        params = {
            "username": username,
            "password": password
        }
        res_get = requests.get(f"http://127.0.0.1:9991/login",json = params)
        arr = res_get.text[-6:-2]
        # print(arr)
        if arr == "登錄成功":
            print(res_get.text)
            print("--------------------恭喜進(jìn)入用戶表操作頁面--------------------")
            while True:
                flag1 = input("輸入：1、查詢用戶信息頁面\t\t2、修改用戶信息頁面\t\t3、刪除用戶信息頁面\t\t4、新增用戶信息頁面\t\t5、退出用戶表操作頁面")
                print(user_table(int(flag1)))
                if flag1 == '5':
                    break
        else:
            print(res_get.text)
    elif flag == '2':
        #post
        print("--------注冊(cè)頁面----------")
        username = input("請(qǐng)輸入用戶名：")
        password = input("請(qǐng)輸入密碼：")
        params = {
            "username": username,
            "password": password
        }
        res_post = requests.post("http://127.0.0.1:9991/register", json=params)
        print(res_post.text)
    else:
        conn = sqlite3.connect("../text2.db", check_same_thread=False)
        conn.close()
        break