15.3.MSSQL

import pymssql

Windows 10
Python 3.8.8 @ MSC v.1928 64 bit (AMD64)
Latest build date 2021.06.30
pymssql version:  2.2.1

SQL 操作

create a connection

似乎 pymssql 库不能使用 DATABASE_URL format。

# Connect to an existing database
try:
    # DATABASE_URL format
    db_url = f"mssql://{user}:{password}@{host}:{port}/{database}"
    conn = pymssql.connect(db_url)
except Exception as e:
    # key parameter format
    conn = pymssql.connect(host=host, port=port,
                           user=user, password=password,
                           database=database)

open a cursor to execute SQL

# Open a cursor to perform database operations
cur = conn.cursor()

# Execute a command
cur.execute("SELECT @@version")
# cur.executemany()

cur.fetchall()
# fetchone
# fetchmany

[('Microsoft SQL Server 2016 (RTM) - 13.0.1601.5 (X64) \n\tApr 29 2016
23:23:58 \n\tCopyright (c) Microsoft Corporation\n\tDeveloper Edition
(64-bit) on Windows Server 2016 Datacenter 6.3 <X64> (Build 14393: )
(Hypervisor)\n',)]

cur.execute("SELECT current_user;")
cur.fetchall()

[('dbo',)]

commit and close connection

# Make the changes to the database persistent
conn.commit()

# Close communication with the database
cur.close()
conn.close()

检查 server SQL 数据库是否可用

from pymssql import OperationalError as MsOperationalError


def check_sql_server_connection(host, username, password, database,
                                port="1433"):
    """Check for SQL server database available.

    Defect: Cannot distinguish specific error messages
            from return result of pymssql library.
            Maybe should use other library to connect SQL server.
    """
    try:
        conn = pymssql.connect(host=host, port=port,
                               user=username, password=password,
                               database=database, charset="utf8", timeout=10)
        cur = conn.cursor()
        cur.execute("select 1;")
        cur.fetchall()
        conn.commit()
        cur.close()
        conn.close()
        return "database available"
    except MsOperationalError as e:
        error_str = e.__str__()
        host_port_error_str = \
            f"Unable to connect: Adaptive Server is unavailable or " \
            f"does not exist ({host}:{port})"
        authentication_error_str = f"Login failed for user '{username}'"
        if host_port_error_str in error_str:
            return "Incorrect IP address or port"
        elif authentication_error_str in error_str:
            return "Incorrect username or password"
        else:
            return "An error occurred while connecting to the database"
    except Exception as e:
        return "An error occurred while connecting to the database"