Data Science. SQL hub on Telegram: “📌Подключаем SQLAlchemy к существующей базой дан…”

📌Подключаем SQLAlchemy к существующей базой данных SQL Server в Python

Создание ORM-модели SQLAlchemy из существующей базы данных SQL Server.

📖Библиотеки

* SQLAlchemy
* pyodbc

Вы можете установить эти библиотеки с помощью команды pip в терминале.

pip install sqlalchemy

pip install pyodbc

Давайте писать!

Во-первых, нам нужно создать базовый класс для нашей модели. Создайте файл с именем base.py и добавьте в него следующий код:

from sqlalchemy.ext.declarative import declarative_base

Base = declarative_base()

Теперь нам нужно сопоставить существующую таблицу с нашим классом. В данном случае мы сопоставим таблицу с двумя колонками с нашей моделью. Таблица представляет собой базу клиентов с полями кода и именем. Для этого мы создадим файл customer.py и добавим в него следующий код:

import sqlalchemy as sa
from base import Base # Reference to the previously created class

class Customer(Base):
    __tablename__ = 'customer' # Name of the table in the database
    customerid = sa.Column('customerid',sa.String(20), primary_key=True)
    customername = sa.Column('customername',sa.String(20))

После того как мы создали базовый класс и класс отображения таблиц, пришло время создать модель. Для этого мы создадим отдельный файл dbmodel.py и импортируем класс таблиц клиентов, который мы только что создали, чтобы выполнять в нем запросы.

from sqlalchemy import create_engine, select
from sqlalchemy.orm import sessionmaker
from customer import Customer # Customer map class

class dbmodel(object):
    db_config = ''
    engine = object

    def __init__(self, conn_string) -> None:
        self.engine = create_engine(conn_string)

    # Function to get the name of a customer by id
    def get_customer_name_by_id(self, customer_id):
        Session = sessionmaker(bind=self.engine, future=True)
        session = Session()
        result = customer_id
        try:
            customer_row = session.execute(
                select(
                    Customer
                ).where(
                    Customer.customerid == customer_id
                )
            ).scalar_one_or_none()
            
            if customer_row is not None:
                result = customer_row.customername

            return result
        except Exception as ex:
            print(ex.args)

Обратите внимание, что в коде мы передаем строку подключения в качестве параметра. Эта строка подключения немного отличается от той, что нужна для SQLite, и мы определим ее в файле run.py, где мы соберем все вместе и запустим тест. Добавим в файл следующий код:

import sys
import dbmodel as db

def main(argv):
    user = 'youruser'
    password = 'yourpassword'
    server = 'yourhost'
    dbname = 'yourdbname'
    
    # SQL Server connection string
    connString = 'mssql+pyodbc://{0}:{1}@{2}/{3}?driver=SQL Server'.format(user, password, server, dbname)

    accountToGet = 'xxx'

    dbSQL = db.dbmodel(connString)
    customer_name = dbSQL.get_customer_name_by_id(accountToGet)
    print(customer_name)

if __name__ == '__main__':
    main(sys.argv[1:])

@sqlhub