python - 如何使用 SQLAlchemy 在 MySQL 中创建数据库？

Question

我需要使用 SQLAlchemy 在 MySQL 中创建一个数据库，如果它已经存在，我可以连接到它，但如果它不存在，我希望能够创建它。这些是我的表：

    #def __init__(self):
Base = declarative_base()

class utente(Base):
    __tablename__="utente"
    utente_id=Column(Integer,primary_key=True)
    nome_utente=Column(Unicode(20))
    ruolo=Column(String(10))
    MetaData.create_all()

    def __repr(self):
        return "utente: {0}, {1}, id: {2}".format(self.ruolo,self.nome_utente,self.utente_id)
    
    
class dbmmas(Base):
    
    __tablename__="dbmmas"
    db_id=Column(Integer,primary_key=True,autoincrement=True)
    nome_db=Column(String(10))
    censimento=Column(Integer)
    versione=Column(Integer)
    ins_data=Column(DateTime)
    mod_data=Column(DateTime)
    ins_utente=Column(Integer)
    mod_utente=Column(Integer)
    MetaData.create_all()

    def __repr(self):
        return "dbmmas: {0}, censimento {1}, versione {2}".format(self.nome_db,self.censimento,self.versione)    
    
class funzione(Base):
    __tablename__="funzione"
    funzione_id=Column(Integer,primary_key=True,autoincrement=True)
    categoria=Column(String(10))
    nome=Column(String(20))
    def __repr__(self):
        return "funzione:{0},categoria:{1},id:{2} ".format(self.nome,self.categoria,self.funzione_id)
    
class profilo(Base):
    __tablename__="rel_utente_funzione" 
    utente_id=Column(Integer,primary_key=True)
    funzione_id=Column(Integer,primary_key=True)
    amministratore=Column(Integer)
    MetaData.create_all()
    
    def __repr(self):
        l=lambda x: "amministratore" if x==1 else "generico"
        return "profilo per utente_id:{0}, tipo: {1}, funzione_id: {2}".format(self.utente_id,l(self.amministratore),self.funzione_id)    
    
class aree(Base):
    __tablename__="rel_utente_zona"
    UTB_id=Column(String(10), primary_key=True) # "in realta' si tratta della seatureSignature della feature sullo shapefile"
    utente_id=Column(Integer, primary_key=True)
    amministratore=Column(Integer)
    MetaData.create_all()
    def __repr(self):
        l=lambda x: "amministratore" if x==1 else "generico"
        return "zona: {0}, pe utente_id:{1}, {2}".format(self.UTB_id,self.utente_id,l(self.amministratore))
    
class rel_utente_dbmmas(Base):
    __tablename__="rel_utente_dbmmas"
    utente_id=Column(Integer,primary_key=True)
    db_id=Column(Integer,primary_key=True)
    amministratore=(Integer)
    MetaData.create_all()
    def __repr(self):
        l=lambda x: "amministratore" if x==1 else "generico"
        return "dbregistrato: {0} per l'utente{1} {2}".format(self.db_id,self.utente_id,l(self.amministratore))

Answer 1

要创建mysql数据库，您只需连接到服务器并创建数据库：

import sqlalchemy
engine = sqlalchemy.create_engine('mysql://user:password@server') # connect to server
engine.execute("CREATE DATABASE dbname") #create db
engine.execute("USE dbname") # select new db
# use the new db
# continue with your work...

当然，您的用户必须具有创建数据库的权限。

Answer 2

您可以为此使用SQLAlchemy-Utils。

pip install sqlalchemy-utils

然后你可以做类似的事情

from sqlalchemy_utils import create_database, database_exists
url = 'mysql://{0}:{1}@{2}:{3}'.format(user, pass, host, port)
if not database_exists(url):
    create_database(url)

我在这里找到了答案，它对我帮助很大。

Answer 3

我不知道规范的方法是什么，但这里有一种方法可以通过检查数据库列表来检查数据库是否存在，如果不存在则创建它。

from sqlalchemy import create_engine

# This engine just used to query for list of databases
mysql_engine = create_engine('mysql://{0}:{1}@{2}:{3}'.format(user, pass, host, port))

# Query for existing databases
existing_databases = mysql_engine.execute("SHOW DATABASES;")
# Results are a list of single item tuples, so unpack each tuple
existing_databases = [d[0] for d in existing_databases]

# Create database if not exists
if database not in existing_databases:
    mysql_engine.execute("CREATE DATABASE {0}".format(database))
    print("Created database {0}".format(database))

# Go ahead and use this engine
db_engine = create_engine('mysql://{0}:{1}@{2}:{3}/{4}'.format(user, pass, host, port, db))

如果您不需要知道数据库是否已创建，这是另一种方法。

from sqlalchemy import create_engine

# This engine just used to query for list of databases
mysql_engine = create_engine('mysql://{0}:{1}@{2}:{3}'.format(user, pass, host, port))

# Query for existing databases
mysql_engine.execute("CREATE DATABASE IF NOT EXISTS {0} ".format(database))

# Go ahead and use this engine
db_engine = create_engine('mysql://{0}:{1}@{2}:{3}/{4}'.format(user, pass, host, port, db))

Answer 4

CREATE DATABASE IF NOT EXISTS dbName;

Answer 5

建议使用with：

from sqlalchemy import create_engine

username = ''
password = ''
host = 'localhost'
port = 3306
DB_NAME = 'db_name'

engine = create_engine(f"mysql://{username}:{password}@{host}:{port}")

with engine.connect() as conn:
    # Do not substitute user-supplied database names here.
    conn.execute(f"CREATE DATABASE IF NOT EXISTS {DB_NAME}")

Answer 6

mysqlclient在基准测试中似乎比PyMySQL快 10 倍，请参阅：MySQLdb、mysqlclient 和 MySQL 连接器/Python 之间有什么区别？.

然而，为什么不为 Python 使用 Python-ready 包，至少，如果不是每秒查询时间呢？以下链接建议使用PyMySQL，例如：

• 使用 SQLAlchemy 访问 MySQL 而不会遇到令人沮丧的库安装问题
• 如何使用 Python+SQLAlchemy 远程连接 MySQL 数据库？.

Python 包：

使用 pip 安装，最好放在“requirements.txt”中：

• PyMySQL
• SQL炼金术

同样，如果它是关于查询的最佳速度，请使用 mysqlclient 包。然后你需要安装一个额外的 Linux 包sudo apt-get install libmysqlclient-dev。

import陈述

只需要一个：

import sqlalchemy

连接字符串 (= db_url)

以{dialect/DBAPI} +{driver}开头的连接字符串：

db_url =mysql+pymysql://

其中pymysql代表使用的 Python 包“PyMySQL”作为驱动程序。

同样，如果它是关于查询的最佳速度，请使用 mysqlclient 包。那么此时你需要mysql+msqldb://。

对于远程连接，您需要添加到连接字符串：

• 主持人
• 用户
• 密码
• 数据库
• 端口（仅当它不是标准 3306 时的端口）

您可以使用多种方法创建 db_url。不要直接在字符串中写入用户名和密码以及最多任何其他变量值，以避免可能的攻击：

• sqlalchemy.engine.URL.create()或使用.url.URL，请参阅从 Cloud Functions 连接到 Cloud SQL中的示例或自动添加?后缀的示例，例如，在为 mssql+pyodbc 构建连接 URL?driver=SQL+Server的字符串末尾使用 sqlalchemy.engine.url.URL
• f"""...{my_var}..."""
• """...{my_var}...""".format(my_var=xyz_var)
• ...

没有 SQLAlchemy 的 url 助手的示例：

db_url = "{dialect}+{driver}://{user}:{password}@{host}:{port}/{database}".format(

或者：

db_url = "{dialect}+{driver}://{user}:{password}@{host}/{database}?host={host}?port={port}".format(
         dialect = 'mysql',
         driver = 'pymysql',
         username=db_user,
         password=db_pass,
         database=db_name,
         host=db_host,
         port=db_port
     )

其他发动机配置

有关其他连接驱动程序、方言和方法，请参阅SQLAlchemy 1.4 文档 - 引擎配置

如果不存在则创建数据库

请参阅如何使用 SQLAlchemy 创建新数据库？.

engine = sqlalchemy.create_engine(db_url)
if not sqlalchemy.database_exists(engine.url):
    create_database(engine.url)
with engine.connect() as conn:
    conn.execute("commit")
    conn.execute("create database test")