使用SQL数据库在Python中进行CRUD操作

CREATE DATABASE Test
GO

USE Test
GO

CREATE TABLE Employee(Id int PRIMARY KEY IDENTITY(1,1),	Name varchar(255) NULL,	Age int NULL)

# SQL Server Database Connection Properties
DATABASE_CONFIG ={
    'Driver': 'SQL Server',
    'Server': 'DESKTOP-QQ95JAD',
    'Database': 'Test',
    'UID': 'sa',
    'Password': 'xxxxxxx'
    }

import pypyodbc
import config

# Return the sql connection 
def getConnection():
     connection = pypyodbc.connect("Driver= {"+config.DATABASE_CONFIG["Driver"]+"} ;
     Server=" + config.DATABASE_CONFIG["Server"] + ";
     Database=" + config.DATABASE_CONFIG["Database"] + ";
     uid=" + config.DATABASE_CONFIG["UID"] + ";pwd=" + config.DATABASE_CONFIG["Password"])
     return connection

现在，让我们从开始执行的地方装饰main函数。如您所见，以下代码是PythonCRUD.py文件，当我们在Visual Studio中使用与项目名称相似的名称创建项目时，将自动创建该文件。在这里，首先，我们将导入所有必需的模块，如用于数据库连接的pypyodbc和使用SQL Server的操作、用于根据上面提供的配置获得连接的db_connection。以及另外四个文件，分别用于执行CRUD操作的读取、创建、更新和删除。

import pypyodbc
import db_connection as dbConn
from read import Read
from create import Create
from update import Update
from delete import Delete

def main():
    print('Available Options: C=Create, R=Read, U=Update, D=Delete ')
    choice = input('Choose your option = ')

    if choice == 'C':
        createObj=Create()
        createObj.func_CreateData()
    elif choice == 'R':
        readObj =  Read()
        readObj.func_ReadData()
    elif choice == 'U':
        updateObj = Update()
        updateObj.func_UpdateData()
    elif choice == 'D':
        deleteObj = Delete()
        deleteObj.func_DeleteData()
    else:
        print('Wrong choice, You are going exist.')

# Call the main function
main()

import db_connection as dbConn

class Create:
    def func_CreateData(self):

        # Get the sql connection
        connection = dbConn.getConnection()
                
        name = input('Enter Name = ')
        age = input('Enter Age = ')

        try:
           query = "Insert Into Employee(Name, Age) Values(?,?)" 
           cursor = connection.cursor()

           # Execute the sql query
           cursor.execute(query, [name, age])

           # Commit the data
           connection.commit()
           print('Data Saved Successfully')

        except:
             print('Something wrong, please check')

        finally:
           # Close the connection
           connection.close()

让我们运行项目，您将获得类似于下面的窗口。在这里，您将获得可供选择的选项[代码将在本文的底部提供]。因此，首先选择您的选项作为C。这意味着，您将创建一个新记录。

import db_connection as dbConn

class Read:
    def func_ReadData(self):   
        # Get the sql connection
        connection = dbConn.getConnection()
        cursor = connection.cursor()

        # Execute the sql query
        cursor.execute('Select * from Employee')

        # Print the data
        for row in cursor:
            print('row = %r' % (row,))

在运行项目时，只需选择R作为读取数据的选项，然后按Enter。您将获得如下输出：

import db_connection as dbConn;

class Update:
    def func_UpdateData(self):
        # Get the SQL connection
        connection = dbConn.getConnection()

        id = input('Enter Employee Id = ')
    
        try:
           # Fetch the data which needs to be updated
           sql = "Select * From Employee Where Id = ?" 
           cursor = connection.cursor()
           cursor.execute(sql, [id])
           item = cursor.fetchone()
           print('Data Fetched for Id = ', id)
           print('ID\t\t Name\t\t\t Age')
           print('-------------------------------------------')       
           print(' {}\t\t {} \t\t\t{} '.format(item[0], item[1], item[2]))
           print('-------------------------------------------')
           print('Enter New Data To Update Employee Record ')

           name = input('Enter New Name = ')
           age = input('Enter New Age = ')
           query = "Update Employee Set Name = ?, Age =? Where Id =?" 
       
           # Execute the update query
           cursor.execute(query, [name, age, id])
           connection.commit()
           print('Data Updated Successfully')

        except:
             print('Something wrong, please check')

        finally:
           # Close the connection
           connection.close()

这次在运行项目时，只需选择U作为更新记录的选项。首先，它将要求输入要更新的员工ID，并显示该员工的数据。接下来，它将要求输入该雇员的姓名和年龄的更新数据并进行更新。成功更新后，将显示确认消息。

import db_connection as dbConn;

class Delete:
    def func_DeleteData(self):
        # Get the SQL connection
        connection = dbConn.getConnection()

        id = input('Enter Employee Id = ')
    
        try:
           # Get record which needs to be deleted
           sql = "Select * From Employee Where Id = ?" 
           cursor = connection.cursor()
           cursor.execute(sql, [id])
           item = cursor.fetchone()
           print('Data Fetched for Id = ', id)
           print('ID\t\t Name\t\t\t Age')
           print('-------------------------------------------')       
           print(' {}\t\t {} \t\t\t{} '.format(item[0], item[1], item[2]))
           print('-------------------------------------------')
           confirm = input('Are you sure to delete this record (Y/N)?')

           # Delete after confirmation
           if confirm == 'Y':
               deleteQuery = "Delete From Employee Where Id = ?"
               cursor.execute(deleteQuery,[id])
               connection.commit()
               print('Data deleted successfully!')
           else:
                print('Wrong Entry')
        except:
            print('Something wrong, please check')
        finally:
            connection.close()

运行项目，然后选择最后一个选项D作为删除记录，并提供需要删除的员工ID。提供员工编号后，首先将显示记录数据并要求确认。如果提供Y作为确认，则它将删除记录并显示确认消息，表明数据已成功删除。