发帖
翼度科技»论坛 › 编程开发 › python › 查看内容

Python史上最全种类数据库操作方法，你能想到的数据库类型都在里面！甚至还

包大仁

8

主题

8

帖子

24

积分

新手上路

Rank: 1

积分
24
显示全部楼层
本文将详细探讨如何在Python中连接全种类数据库以及实现相应的CRUD（创建，读取，更新，删除）操作。我们将逐一解析连接MySQL，SQL Server，Oracle，PostgreSQL，MongoDB，SQLite，DB2，Redis，Cassandra，Microsoft Access，ElasticSearch，Neo4j，InfluxDB，Snowflake，Amazon DynamoDB，Microsoft Azure CosMos DB数据库的方法，并演示相应的CRUD操作。
MySQL

连接数据库

Python可以使用mysql-connector-python库连接MySQL数据库：
  1. import mysql.connector
  3. conn = mysql.connector.connect(user='username', password='password', host='127.0.0.1', database='my_database')
  4. print("Opened MySQL database successfully")
  5. conn.close()
复制代码
CRUD操作

接下来，我们将展示在MySQL中如何进行基本的CRUD操作。
创建（Create）
  1. conn = mysql.connector.connect(user='username', password='password', host='127.0.0.1', database='my_database')
  3. cursor = conn.cursor()
  4. cursor.execute("CREATE TABLE Employees (ID INT PRIMARY KEY NOT NULL, NAME TEXT NOT NULL, AGE INT, ADDRESS CHAR(50), SALARY REAL)")
  5. print("Table created successfully")
  7. conn.close()
复制代码
读取（Retrieve）
  1. conn = mysql.connector.connect(user='username', password='password', host='127.0.0.1', database='my_database')
  3. cursor = conn.cursor()
  4. cursor.execute("SELECT id, name, address, salary from Employees")
  5. rows = cursor.fetchall()
  6. for row in rows:
  7.     print("ID = ", row[0])
  8.     print("NAME = ", row[1])
  9.     print("ADDRESS = ", row[2])
  10.     print("SALARY = ", row[3])
  12. conn.close()
复制代码
更新（Update）
  1. conn = mysql.connector.connect(user='username', password='password', host='127.0.0.1', database='my_database')
  3. cursor = conn.cursor()
  4. cursor.execute("UPDATE Employees set SALARY = 25000.00 where ID = 1")
  5. conn.commit()
  7. print("Total number of rows updated :", cursor.rowcount)
  9. conn.close()
复制代码
删除（Delete）
  1. conn = mysql.connector.connect(user='username', password='password', host='127.0.0.1', database='my_database')
  3. cursor = conn.cursor()
  4. cursor.execute("DELETE from Employees where ID = 1")
  5. conn.commit()
  7. print("Total number of rows deleted :", cursor.rowcount)
  9. conn.close()
复制代码
SQL Server

连接数据库

Python可以使用pyodbc库连接SQL Server数据库：
  1. import pyodbc
  3. conn = pyodbc.connect('DRIVER={SQL Server};SERVER=localhost;DATABASE=my_database;UID=username;PWD=password')
  4. print("Opened SQL Server database successfully")
  5. conn.close()
复制代码
CRUD操作

接下来，我们将展示在SQL Server中如何进行基本的CRUD操作。
创建（Create）
  1. conn = pyodbc.connect('DRIVER={SQL Server};SERVER=localhost;DATABASE=my_database;UID=username;PWD=password')
  3. cursor = conn.cursor()
  4. cursor.execute("CREATE TABLE Employees (ID INT PRIMARY KEY NOT NULL, NAME VARCHAR(20) NOT NULL, AGE INT, ADDRESS CHAR(50), SALARY REAL)")
  5. conn.commit()
  6. print("Table created successfully")
  8. conn.close()
复制代码
读取（Retrieve）
  1. conn = pyodbc.connect('DRIVER={SQL Server};SERVER=localhost;DATABASE=my_database;UID=username;PWD=password')
  3. cursor = conn.cursor()
  4. cursor.execute("SELECT id, name, address, salary from Employees")
  5. rows = cursor.fetchall()
  6. for row in rows:
  7.     print("ID = ", row[0])
  8.     print("NAME = ", row[1])
  9.     print("ADDRESS = ", row[2])
  10.     print("SALARY = ", row[3])
  12. conn.close()
复制代码
更新（Update）
  1. conn = pyodbc.connect('DRIVER={SQL Server};SERVER=localhost;DATABASE=my_database;UID=username;PWD=password')
  3. cursor = conn.cursor()
  4. cursor.execute("UPDATE Employees set SALARY = 25000.00 where ID = 1")
  5. conn.commit()
  7. print("Total number of rows updated :", cursor.rowcount)
  9. conn.close()
复制代码
删除（Delete）
  1. conn = pyodbc.connect('DRIVER={SQL Server};SERVER=localhost;DATABASE=my_database;UID=username;PWD=password')
  3. cursor = conn.cursor()
  4. cursor.execute("DELETE from Employees where ID = 1")
  5. conn.commit()
  7. print("Total number of rows deleted :", cursor.rowcount)
  9. conn.close()
复制代码
Oracle

连接数据库

Python可以使用cx_Oracle库连接Oracle数据库：
  1. import cx_Oracle
  3. dsn_tns = cx_Oracle.makedsn('localhost', '1521', service_name='my_database')
  4. conn = cx_Oracle.connect(user='username', password='password', dsn=dsn_tns)
  5. print("Opened Oracle database successfully")
  6. conn.close()
复制代码
CRUD操作

接下来，我们将展示在Oracle中如何进行基本的CRUD操作。
创建（Create）
  1. dsn_tns = cx_Oracle.makedsn('localhost', '1521', service_name='my_database')
  2. conn = cx_Oracle.connect(user='username', password='password', dsn=dsn_tns)
  4. cursor = conn.cursor()
  5. cursor.execute("CREATE TABLE Employees (ID NUMBER(10) NOT NULL PRIMARY KEY, NAME VARCHAR2(20) NOT NULL, AGE NUMBER(3), ADDRESS CHAR(50), SALARY NUMBER(10, 2))")
  6. conn.commit()
  7. print("Table created successfully")
  9. conn.close()
复制代码
读取（Retrieve）
  1. dsn_tns = cx_Oracle.makedsn('localhost', '1521', service_name='my_database')
  2. conn = cx_Oracle.connect(user='username', password='password', dsn=dsn_tns)
  4. cursor = conn.cursor()
  5. cursor.execute("SELECT id, name, address, salary from Employees")
  6. rows = cursor.fetchall()
  7. for row in rows:
  8.     print("ID = ", row[0])
  9.     print("NAME = ", row[1])
  10.     print("ADDRESS = ", row[2])
  11.     print("SALARY = ", row[3])
  13. conn.close()
复制代码
更新（Update）
  1. dsn_tns = cx_Oracle.makedsn('localhost', '1521', service_name='my_database')
  2. conn = cx_Oracle.connect(user='username', password='password', dsn=dsn_tns)
  4. cursor = conn.cursor()
  5. cursor.execute("UPDATE Employees set SALARY = 25000.00 where ID = 1")
  6. conn.commit()
  8. print("Total number of rows updated :", cursor.rowcount)
  10. conn.close()
复制代码
删除（Delete）
  1. dsn_tns = cx_Oracle.makedsn('localhost', '1521', service_name='my_database')
  2. conn = cx_Oracle.connect(user='username', password='password', dsn=dsn_tns)
  4. cursor = conn.cursor()
  5. cursor.execute("DELETE from Employees where ID = 1")
  6. conn.commit()
  8. print("Total number of rows deleted :", cursor.rowcount)
  10. conn.close()
复制代码
PostgreSQL

连接数据库

Python可以使用psycopg2库连接PostgreSQL数据库：
  1. import psycopg2
  3. conn = psycopg2.connect(database="my_database", user="username", password="password", host="127.0.0.1", port="5432")
  4. print("Opened PostgreSQL database successfully")
  5. conn.close()
复制代码
CRUD操作

接下来，我们将展示在PostgreSQL中如何进行基本的CRUD操作。
创建（Create）
  1. conn = psycopg2.connect(database="my_database", user="username", password="password", host="127.0.0.1", port="5432")
  3. cursor = conn.cursor()
  4. cursor.execute('''CREATE TABLE Employees
  5.       (ID INT PRIMARY KEY     NOT NULL,
  6.       NAME           TEXT    NOT NULL,
  7.       AGE            INT     NOT NULL,
  8.       ADDRESS        CHAR(50),
  9.       SALARY         REAL);''')
  10. conn.commit()
  11. print("Table created successfully")
  13. conn.close()
复制代码
读取（Retrieve）
  1. conn = psycopg2.connect(database="my_database", user="username", password="password", host="127.0.0.1", port="5432")
  3. cursor = conn.cursor()
  4. cursor.execute("SELECT id, name, address, salary from Employees")
  5. rows = cursor.fetchall()
  6. for row in rows:
  7.     print("ID = ", row[0])
  8.     print("NAME = ", row[1])
  9.     print("ADDRESS = ", row[2])
  10.     print("SALARY = ", row[3])
  12. conn.close()
复制代码
更新（Update）
  1. conn = psycopg2.connect(database="my_database", user="username", password="password", host="127.0.0.1", port="5432")
  3. cursor = conn.cursor()
  4. cursor.execute("UPDATE Employees set SALARY = 25000.00 where ID = 1")
  5. conn.commit()
  7. print("Total number of rows updated :", cursor.rowcount)
  9. conn.close()
复制代码
删除（Delete）
  1. conn = psycopg2.connect(database="my_database", user="username", password="password", host="127.0.0.1", port="5432")
  3. cursor = conn.cursor()
  4. cursor.execute("DELETE from Employees where ID = 1")
  5. conn.commit()
  7. print("Total number of rows deleted :", cursor.rowcount)
  9. conn.close()
复制代码
MongoDB

连接数据库

Python可以使用pymongo库连接MongoDB数据库：
  1. from pymongo import MongoClient
  3. client = MongoClient("mongodb://localhost:27017/")
  4. db = client["my_database"]
  5. print("Opened MongoDB database successfully")
  6. client.close()
复制代码
CRUD操作

接下来，我们将展示在MongoDB中如何进行基本的CRUD操作。
创建（Create）

在MongoDB中，文档的创建操作通常包含在插入操作中：
  1. client = MongoClient("mongodb://localhost:27017/")
  2. db = client["my_database"]
  4. employees = db["Employees"]
  5. employee = {"id": "1", "name": "John", "age": "30", "address": "New York", "salary": "1000.00"}
  7. employees.insert_one(employee)
  8. print("Document inserted successfully")
  10. client.close()
复制代码
读取（Retrieve）
  1. client = MongoClient("mongodb://localhost:27017/")
  2. db = client["my_database"]
  4. employees = db["Employees"]
  5. cursor = employees.find()
  6. for document in cursor:
  7.     print(document)
  9. client.close()
复制代码
更新（Update）
  1. client = MongoClient("mongodb://localhost:27017/")
  2. db = client["my_database"]
  4. employees = db["Employees"]
  5. query = { "id": "1" }
  6. new_values = { "$set": { "salary": "25000.00" } }
  8. employees.update_one(query, new_values)
  10. print("Document updated successfully")
  12. client.close()
复制代码
删除（Delete）
  1. client = MongoClient("mongodb://localhost:27017/")
  2. db = client["my_database"]
  4. employees = db["Employees"]
  5. query = { "id": "1" }
  7. employees.delete_one(query)
  9. print("Document deleted successfully")
  11. client.close()
复制代码
SQLite

连接数据库

Python使用sqlite3库连接SQLite数据库：
  1. import sqlite3
  3. conn = sqlite3.connect('my_database.db')
  4. print("Opened SQLite database successfully")
  5. conn.close()
复制代码
CRUD操作

接下来，我们将展示在SQLite中如何进行基本的CRUD操作。
创建（Create）
  1. conn = sqlite3.connect('my_database.db')
  3. cursor = conn.cursor()
  4. cursor.execute('''CREATE TABLE Employees
  5.       (ID INT PRIMARY KEY     NOT NULL,
  6.       NAME           TEXT    NOT NULL,
  7.       AGE            INT     NOT NULL,
  8.       ADDRESS        CHAR(50),
  9.       SALARY         REAL);''')
  10. conn.commit()
  11. print("Table created successfully")
  13. conn.close()
复制代码
读取（Retrieve）
  1. conn = sqlite3.connect('my_database.db')
  3. cursor = conn.cursor()
  4. cursor.execute("SELECT id, name, address, salary from Employees")
  5. rows = cursor.fetchall()
  6. for row in rows:
  7.     print("ID = ", row[0])
  8.     print("NAME = ", row[1])
  9.     print("ADDRESS = ", row[2])
  10.     print("SALARY = ", row[3])
  12. conn.close()
复制代码
更新（Update）
  1. conn = sqlite3.connect('my_database.db')
  3. cursor = conn.cursor()
  4. cursor.execute("UPDATE Employees set SALARY = 25000.00 where ID = 1")
  5. conn.commit()
  7. print("Total number of rows updated :", cursor.rowcount)
  9. conn.close()
复制代码
删除（Delete）
  1. conn = sqlite3.connect('my_database.db')
  3. cursor = conn.cursor()
  4. cursor.execute("DELETE from Employees where ID = 1")
  5. conn.commit()
  7. print("Total number of rows deleted :", cursor.rowcount)
  9. conn.close()
复制代码
DB2

连接数据库

Python可以使用ibm_db库连接DB2数据库：
  1. import ibm_db
  3. dsn = (
  4.     "DRIVER={{IBM DB2 ODBC DRIVER}};"
  5.     "DATABASE=my_database;"
  6.     "HOSTNAME=127.0.0.1;"
  7.     "PORT=50000;"
  8.     "PROTOCOL=TCPIP;"
  9.     "UID=username;"
  10.     "PWD=password;"
  11. )
  12. conn = ibm_db.connect(dsn, "", "")
  13. print("Opened DB2 database successfully")
  14. ibm_db.close(conn)
复制代码
CRUD操作

接下来，我们将展示在DB2中如何进行基本的CRUD操作。
创建（Create）
  1. conn = ibm_db.connect(dsn, "", "")
  3. sql = '''CREATE TABLE Employees
  4.       (ID INT PRIMARY KEY     NOT NULL,
  5.       NAME           VARCHAR(20)    NOT NULL,
  6.       AGE            INT     NOT NULL,
  7.       ADDRESS        CHAR(50),
  8.       SALARY         DECIMAL(9, 2));'''
  9. stmt = ibm_db.exec_immediate(conn, sql)
  10. print("Table created successfully")
  12. ibm_db.close(conn)
复制代码
读取（Retrieve）
  1. conn = ibm_db.connect(dsn, "", "")
  3. sql = "SELECT id, name, address, salary from Employees"
  4. stmt = ibm_db.exec_immediate(conn, sql)
  5. while ibm_db.fetch_row(stmt):
  6.     print("ID = ", ibm_db.result(stmt, "ID"))
  7.     print("NAME = ", ibm_db.result(stmt, "NAME"))
  8.     print("ADDRESS = ", ibm_db.result(stmt, "ADDRESS"))
  9.     print("SALARY = ", ibm_db.result(stmt, "SALARY"))
  11. ibm_db.close(conn)
复制代码
更新（Update）
  1. conn = ibm_db.connect(dsn, "", "")
  3. sql = "UPDATE Employees set SALARY = 25000.00 where ID = 1"
  4. stmt = ibm_db.exec_immediate(conn, sql)
  5. ibm_db.commit(conn)
  7. print("Total number of rows updated :", ibm_db.num_rows(stmt))
  9. ibm_db.close(conn)
复制代码
删除（Delete）
  1. conn = ibm_db.connect(dsn, "", "")
  3. sql = "DELETE from Employees where ID = 1"
  4. stmt = ibm_db.exec_immediate(conn, sql)
  5. ibm_db.commit(conn)
  7. print("Total number of rows deleted :", ibm_db.num_rows(stmt))
  9. ibm_db.close(conn)
复制代码
Microsoft Access

连接数据库

Python可以使用pyodbc库连接Microsoft Access数据库：
  1. import pyodbc
  3. conn_str = (
  4.     r'DRIVER={Microsoft Access Driver (*.mdb, *.accdb)};'
  5.     r'DBQ=path_to_your_access_file.accdb;'
  6. )
  7. conn = pyodbc.connect(conn_str)
  8. print("Opened Access database successfully")
  9. conn.close()
复制代码
CRUD操作

接下来，我们将展示在Access中如何进行基本的CRUD操作。
创建（Create）
  1. conn = pyodbc.connect(conn_str)
  3. cursor = conn.cursor()
  4. cursor.execute('''CREATE TABLE Employees
  5.       (ID INT PRIMARY KEY     NOT NULL,
  6.       NAME           TEXT    NOT NULL,
  7.       AGE            INT     NOT NULL,
  8.       ADDRESS        CHAR(50),
  9.       SALARY         DECIMAL(9, 2));''')
  10. conn.commit()
  11. print("Table created successfully")
  13. conn.close()
复制代码
读取（Retrieve）
  1. conn = pyodbc.connect(conn_str)
  3. cursor = conn.cursor()
  4. cursor.execute("SELECT id, name, address, salary from Employees")
  5. rows = cursor.fetchall()
  6. for row in rows:
  7.     print("ID = ", row[0])
  8.     print("NAME = ", row[1])
  9.     print("ADDRESS = ", row[2])
  10.     print("SALARY = ", row[3])
  12. conn.close()
复制代码
更新（Update）
  1. conn = pyodbc.connect(conn_str)
  3. cursor = conn.cursor()
  4. cursor.execute("UPDATE Employees set SALARY = 25000.00 where ID = 1")
  5. conn.commit()
  7. print("Total number of rows updated :", cursor.rowcount)
  9. conn.close()
复制代码
删除（Delete）
  1. conn = pyodbc.connect(conn_str)
  3. cursor = conn.cursor()
  4. cursor.execute("DELETE from Employees where ID = 1")
  5. conn.commit()
  7. print("Total number of rows deleted :", cursor.rowcount)
  9. conn.close()
复制代码
Cassandra

连接数据库

Python可以使用cassandra-driver库连接Cassandra数据库：
  1. from cassandra.cluster import Cluster
  3. cluster = Cluster(['127.0.0.1'])
  4. session = cluster.connect('my_keyspace')
  5. print("Opened Cassandra database successfully")
  6. cluster.shutdown()
复制代码
CRUD操作

接下来，我们将展示在Cassandra中如何进行基本的CRUD操作。
创建（Create）
  1. cluster = Cluster(['127.0.0.1'])
  2. session = cluster.connect('my_keyspace')
  4. session.execute("""
  5.     CREATE TABLE Employees (
  6.         id int PRIMARY KEY,
  7.         name text,
  8.         age int,
  9.         address text,
  10.         salary decimal
  11.     )
  12. """)
  13. print("Table created successfully")
  15. cluster.shutdown()
复制代码
读取（Retrieve）
  1. cluster = Cluster(['127.0.0.1'])
  2. session = cluster.connect('my_keyspace')
  4. rows = session.execute('SELECT id, name, address, salary FROM Employees')
  5. for row in rows:
  6.     print("ID = ", row.id)
  7.     print("NAME = ", row.name)
  8.     print("ADDRESS = ", row.address)
  9.     print("SALARY = ", row.salary)
  11. cluster.shutdown()
复制代码
更新（Update）
  1. cluster = Cluster(['127.0.0.1'])
  2. session = cluster.connect('my_keyspace')
  4. session.execute("UPDATE Employees SET salary = 25000.00 WHERE id = 1")
  5. print("Row updated successfully")
  7. cluster.shutdown()
复制代码
删除（Delete）
  1. cluster = Cluster(['127.0.0.1'])
  2. session = cluster.connect('my_keyspace')
  4. session.execute("DELETE FROM Employees WHERE id = 1")
  5. print("Row deleted successfully")
  7. cluster.shutdown()
复制代码
Redis

连接数据库

Python可以使用redis-py库连接Redis数据库：
  1. import redis
  3. r = redis.Redis(host='localhost', port=6379, db=0)
  4. print("Opened Redis database successfully")
复制代码
CRUD操作

接下来，我们将展示在Redis中如何进行基本的CRUD操作。
创建（Create）
  1. r = redis.Redis(host='localhost', port=6379, db=0)
  3. r.set('employee:1:name', 'John')
  4. r.set('employee:1:age', '30')
  5. r.set('employee:1:address', 'New York')
  6. r.set('employee:1:salary', '1000.00')
  8. print("Keys created successfully")
复制代码
读取（Retrieve）
  1. r = redis.Redis(host='localhost', port=6379, db=0)
  3. print("NAME = ", r.get('employee:1:name').decode('utf-8'))
  4. print("AGE = ", r.get('employee:1:age').decode('utf-8'))
  5. print("ADDRESS = ", r.get('employee:1:address').decode('utf-8'))
  6. print("SALARY = ", r.get('employee:1:salary').decode('utf-8'))
复制代码
更新（Update）
  1. r = redis.Redis(host='localhost', port=6379, db=0)
  3. r.set('employee:1:salary', '25000.00')
  5. print("Key updated successfully")
复制代码
删除（Delete）
  1. r = redis.Redis(host='localhost', port=6379, db=0)
  3. r.delete('employee:1:name', 'employee:1:age', 'employee:1:address', 'employee:1:salary')
  5. print("Keys deleted successfully")
复制代码
ElasticSearch

连接数据库

Python可以使用elasticsearch库连接ElasticSearch数据库：
  1. from elasticsearch import Elasticsearch
  3. es = Elasticsearch([{'host': 'localhost', 'port': 9200}])
  4. print("Opened ElasticSearch database successfully")
复制代码
CRUD操作

接下来，我们将展示在ElasticSearch中如何进行基本的CRUD操作。
创建（Create）
  1. es = Elasticsearch([{'host': 'localhost', 'port': 9200}])
  3. employee = {
  4.     'name': 'John',
  5.     'age': 30,
  6.     'address': 'New York',
  7.     'salary': 1000.00
  8. }
  9. res = es.index(index='employees', doc_type='employee', id=1, body=employee)
  11. print("Document created successfully")
复制代码
读取（Retrieve）
  1. es = Elasticsearch([{'host': 'localhost', 'port': 9200}])
  3. res = es.get(index='employees', doc_type='employee', id=1)
  4. print("Document details:")
  5. for field, details in res['_source'].items():
  6.     print(f"{field.upper()} = ", details)
复制代码
更新（Update）
  1. es = Elasticsearch([{'host': 'localhost', 'port': 9200}])
  3. res = es.update(index='employees', doc_type='employee', id=1, body={
  4.     'doc': {
  5.         'salary': 25000.00
  6.     }
  7. })
  9. print("Document updated successfully")
复制代码
删除（Delete）
  1. es = Elasticsearch([{'host': 'localhost', 'port': 9200}])
  3. res = es.delete(index='employees', doc_type='employee', id=1)
  5. print("Document deleted successfully")
复制代码
Neo4j

连接数据库

Python可以使用neo4j库连接Neo4j数据库：
  1. from neo4j import GraphDatabase
  3. driver = GraphDatabase.driver("bolt://localhost:7687", auth=("neo4j", "password"))
  4. print("Opened Neo4j database successfully")
  5. driver.close()
复制代码
CRUD操作

接下来，我们将展示在Neo4j中如何进行基本的CRUD操作。
创建（Create）
  1. driver = GraphDatabase.driver("bolt://localhost:7687", auth=("neo4j", "password"))
  3. with driver.session() as session:
  4.     session.run("CREATE (:Employee {id: 1, name: 'John', age: 30, address: 'New York', salary: 1000.00})")
  6. print("Node created successfully")
  8. driver.close()
复制代码
读取（Retrieve）
  1. driver = GraphDatabase.driver("bolt://localhost:7687", auth=("neo4j", "password"))
  3. with driver.session() as session:
  4.     result = session.run("MATCH (n:Employee) WHERE n.id = 1 RETURN n")
  5.     for record in result:
  6.         print("ID = ", record["n"]["id"])
  7.         print("NAME = ", record["n"]["name"])
  8.         print("ADDRESS = ", record["n"]["address"])
  9.         print("SALARY = ", record["n"]["salary"])
  11. driver.close()
复制代码
更新（Update）
  1. driver = GraphDatabase.driver("bolt://localhost:7687", auth=("neo4j", "password"))
  3. with driver.session() as session:
  4.     session.run("MATCH (n:Employee) WHERE n.id = 1 SET n.salary = 25000.00")
  6. print("Node updated successfully")
  8. driver.close()
复制代码
删除（Delete）
  1. driver = GraphDatabase.driver("bolt://localhost:7687", auth=("neo4j", "password"))
  3. with driver.session() as session:
  4.     session.run("MATCH (n:Employee) WHERE n.id = 1 DETACH DELETE n")
  6. print("Node deleted successfully")
  8. driver.close()
复制代码
InfluxDB

连接数据库

Python可以使用InfluxDB-Python库连接InfluxDB数据库：
  1. from influxdb import InfluxDBClient
  3. client = InfluxDBClient(host='localhost', port=8086)
  4. print("Opened InfluxDB database successfully")
  5. client.close()
复制代码
CRUD操作

接下来，我们将展示在InfluxDB中如何进行基本的CRUD操作。
创建（Create）
  1. client = InfluxDBClient(host='localhost', port=8086)
  3. json_body = [
  4.     {
  5.         "measurement": "employees",
  6.         "tags": {
  7.             "id": "1"
  8.         },
  9.         "fields": {
  10.             "name": "John",
  11.             "age": 30,
  12.             "address": "New York",
  13.             "salary": 1000.00
  14.         }
  15.     }
  16. ]
  18. client.write_points(json_body)
  20. print("Point created successfully")
  22. client.close()
复制代码
读取（Retrieve）
  1. client = InfluxDBClient(host='localhost', port=8086)
  3. result = client.query('SELECT "name", "age", "address", "salary" FROM "employees"')
  5. for point in result.get_points():
  6.     print("ID = ", point['id'])
  7.     print("NAME = ", point['name'])
  8.     print("AGE = ", point['age'])
  9.     print("ADDRESS = ", point['address'])
  10.     print("SALARY = ", point['salary'])
  12. client.close()
复制代码
更新（Update）

InfluxDB的数据模型和其他数据库不同，它没有更新操作。但是你可以通过写入一个相同的数据点（即具有相同的时间戳和标签）并改变字段值，实现类似更新操作的效果。
删除（Delete）

同样，InfluxDB也没有提供删除单个数据点的操作。然而，你可以删除整个系列（即表）或者删除某个时间段的数据。
  1. client = InfluxDBClient(host='localhost', port=8086)
  3. # 删除整个系列
  4. client.query('DROP SERIES FROM "employees"')
  6. # 删除某个时间段的数据
  7. # client.query('DELETE FROM "employees" WHERE time < now() - 1d')
  9. print("Series deleted successfully")
  11. client.close()
复制代码
Snowflake

连接数据库

Python可以使用snowflake-connector-python库连接Snowflake数据库：
  1. from snowflake.connector import connect
  3. con = connect(
  4.     user='username',
  5.     password='password',
  6.     account='account_url',
  7.     warehouse='warehouse',
  8.     database='database',
  9.     schema='schema'
  10. )
  11. print("Opened Snowflake database successfully")
  12. con.close()
复制代码
CRUD操作

接下来，我们将展示在Snowflake中如何进行基本的CRUD操作。
创建（Create）
  1. con = connect(
  2.     user='username',
  3.     password='password',
  4.     account='account_url',
  5.     warehouse='warehouse',
  6.     database='database',
  7.     schema='schema'
  8. )
  10. cur = con.cursor()
  11. cur.execute("""
  12. CREATE TABLE EMPLOYEES (
  13.     ID INT,
  14.     NAME STRING,
  15.     AGE INT,
  16.     ADDRESS STRING,
  17.     SALARY FLOAT
  18. )
  19. """)
  21. cur.execute("""
  22. INSERT INTO EMPLOYEES (ID, NAME, AGE, ADDRESS, SALARY) VALUES
  23. (1, 'John', 30, 'New York', 1000.00)
  24. """)
  26. print("Table created and row inserted successfully")
  28. con.close()
复制代码
读取（Retrieve）
  1. con = connect(
  2.     user='username',
  3.     password='password',
  4.     account='account_url',
  5.     warehouse='warehouse',
  6.     database='database',
  7.     schema='schema'
  8. )
  10. cur = con.cursor()
  11. cur.execute("SELECT * FROM EMPLOYEES WHERE ID = 1")
  13. rows = cur.fetchall()
  15. for row in rows:
  16.     print("ID = ", row[0])
  17.     print("NAME = ", row[1])
  18.     print("AGE = ", row[2])
  19.     print("ADDRESS = ", row[3])
  20.     print("SALARY = ", row[4])
  22. con.close()
复制代码
更新（Update）
  1. con = connect(
  2.     user='username',
  3.     password='password',
  4.     account='account_url',
  5.     warehouse='warehouse',
  6.     database='database',
  7.     schema='schema'
  8. )
  10. cur = con.cursor()
  11. cur.execute("UPDATE EMPLOYEES SET SALARY = 25000.00 WHERE ID = 1")
  13. print("Row updated successfully")
  15. con.close()
复制代码
删除（Delete）
  1. con = connect(
  2.     user='username',
  3.     password='password',
  4.     account='account_url',
  5.     warehouse='warehouse',
  6.     database='database',
  7.     schema='schema'
  8. )
  10. cur = con.cursor()
  11. cur.execute("DELETE FROM EMPLOYEES WHERE ID = 1")
  13. print("Row deleted successfully")
  15. con.close()
复制代码
Amazon DynamoDB

连接数据库

Python可以使用boto3库连接Amazon DynamoDB：
  1. import boto3
  3. dynamodb = boto3.resource('dynamodb', region_name='us-west-2',
  4.                           aws_access_key_id='Your AWS Access Key',
  5.                           aws_secret_access_key='Your AWS Secret Key')
  7. print("Opened DynamoDB successfully")
复制代码
CRUD操作

接下来，我们将展示在DynamoDB中如何进行基本的CRUD操作。
创建（Create）
  1. table = dynamodb.create_table(
  2.     TableName='Employees',
  3.     KeySchema=[
  4.         {
  5.             'AttributeName': 'id',
  6.             'KeyType': 'HASH'
  7.         },
  8.     ],
  9.     AttributeDefinitions=[
  10.         {
  11.             'AttributeName': 'id',
  12.             'AttributeType': 'N'
  13.         },
  14.     ],
  15.     ProvisionedThroughput={
  16.         'ReadCapacityUnits': 5,
  17.         'WriteCapacityUnits': 5
  18.     }
  19. )
  21. table.put_item(
  22.    Item={
  23.         'id': 1,
  24.         'name': 'John',
  25.         'age': 30,
  26.         'address': 'New York',
  27.         'salary': 1000.00
  28.     }
  29. )
  31. print("Table created and item inserted successfully")
复制代码
读取（Retrieve）
  1. table = dynamodb.Table('Employees')
  3. response = table.get_item(
  4.    Key={
  5.         'id': 1,
  6.     }
  7. )
  9. item = response['Item']
  10. print(item)
复制代码
更新（Update）
  1. table = dynamodb.Table('Employees')
  3. table.update_item(
  4.     Key={
  5.         'id': 1,
  6.     },
  7.     UpdateExpression='SET salary = :val1',
  8.     ExpressionAttributeValues={
  9.         ':val1': 25000.00
  10.     }
  11. )
  13. print("Item updated successfully")
复制代码
删除（Delete）
  1. table = dynamodb.Table('Employees')
  3. table.delete_item(
  4.     Key={
  5.         'id': 1,
  6.     }
  7. )
  9. print("Item deleted successfully")
复制代码
Microsoft Azure CosMos DB

连接数据库

Python可以使用azure-cosmos库连接Microsoft Azure CosMos DB：
  1. from azure.cosmos import CosmosClient, PartitionKey, exceptions
  3. url = 'Cosmos DB Account URL'
  4. key = 'Cosmos DB Account Key'
  5. client = CosmosClient(url, credential=key)
  7. database_name = 'testDB'
  8. database = client.get_database_client(database_name)
  10. container_name = 'Employees'
  11. container = database.get_container_client(container_name)
  13. print("Opened CosMos DB successfully")
复制代码
CRUD操作

接下来，我们将展示在CosMos DB中如何进行基本的CRUD操作。
创建（Create）
  1. database = client.create_database_if_not_exists(id=database_name)
  3. container = database.create_container_if_not_exists(
  4.     id=container_name,
  5.     partition_key=PartitionKey(path="/id"),
  6.     offer_throughput=400
  7. )
  9. container.upsert_item({
  10.     'id': '1',
  11.     'name': 'John',
  12.     'age': 30,
  13.     'address': 'New York',
  14.     'salary': 1000.00
  15. })
  17. print("Container created and item upserted successfully")
复制代码
读取（Retrieve）
  1. for item in container.read_all_items():
  2.     print(item)
复制代码
更新（Update）
  1. for item in container.read_all_items():
  2.     if item['id'] == '1':
  3.         item['salary'] = 25000.00
  4.         container.upsert_item(item)
  5.         
  6. print("Item updated successfully")
复制代码
删除（Delete）
  1. for item in container.read_all_items():
  2.     if item['id'] == '1':
  3.         container.delete_item(item, partition_key='1')
  4.         
  5. print("Item deleted successfully")
复制代码
如有帮助，请多关注
个人微信公众号：【Python全视角】
TeahLead_KrisChang，10+年的互联网和人工智能从业经验，10年+技术和业务团队管理经验，同济软件工程本科，复旦工程管理硕士，阿里云认证云服务资深架构师，上亿营收AI产品业务负责人。

来源:https://www.cnblogs.com/xfuture/p/17528203.html
免责声明：由于采集信息均来自互联网，如果侵犯了您的权益，请联系我们【E-Mail:cb@itdo.tech】 我们会及时删除侵权内容，谢谢合作！

上一篇: python3使用PIL添加中文文本水印背景

下一篇: Python教程(2)——开发python常用的IDE

发表于 2023-7-5 15:16:40

举报 回复 使用道具

返回列表 发新帖

Archiver|手机版|小黑屋|翼度科技

Copyright © 2001-2020, Tencent Cloud.　Powered by Discuz! X3.4