SQLAlchemy: Подключение к уже существующим базам данных

Доступ к базам данных с помощью Python — это "простой" ага :) процесс. Python даже предоставляет библиотеку баз данных sqlite, встроенную в основной дистрибутив (начиная с версии 2.5). Самый "модный", по моему мнению, способ доступа к базам данных с помощью Python — использовать сторонний пакет SQLAlchemy. SQLAlchemy — это объектно-реляционное сопоставление (ORM), что означает, что оно использует конструкции SQL и делает их более похожими на целевой язык. В этом случае вы в конечном итоге используете синтаксис Python для выполнения SQL, а не прямого SQL, и вы можете использовать один и тот же код для доступа к нескольким бэкэндам базы данных (если быть осторожным).

В этой статье мы рассмотрим, как использовать SQLAlchemy для подключения к уже существующим базам данных. Если судить по моему опыту, то больше всего тратится времени на работу с базами данных, которые создавали не вы, чем с теми, которые создали вы. Цель этой статьи показать как к ним подключиться.

Буду пояснять на реальном примере присоединения к базе Нижегородского разработчика CAD-системы К3-мебель "ГеоС". У меня есть версия 7.4, которая использует MS Accsess и версия 8.0, которая использует MSSQL Server.

Будем использовать python 3.7

Потребуется установить sqlalchemy-access.

Для начала, разберём примеры от Mike Driscoll. Старая, аж 2010 года, публикация, но актуальная.

Автозагрузка SQLAlchemy

В SQLAlchemy есть два способа определения столбцов базы данных. Первый способ — это длинный путь, в котором вы определяете каждое поле вместе с его типом. Самый простой (или короткий) способ — использовать функцию autoload SQLAlchemy, которая будет анализировать таблицу и извлекать имена полей волшебно-подкапотным способом. Мы начнем с метода автозагрузки, а затем покажем долгий путь в следующем разделе.

Однако, прежде чем мы начнем, необходимо отметить, что существует два метода настройки SQLAlchemy: метод long-hand и декларативный (или "сокращенный") способ. Мы рассмотрим оба пути. Давайте начнем с длинной версии, а затем сделаем это "декларативно".

from sqlalchemy import create_engine, MetaData, Table
from sqlalchemy.orm import mapper, sessionmaker

class Bookmarks(object):
    pass

def loadSession():
    """"""    
    dbPath = 'places.sqlite'
    engine = create_engine('sqlite:///%s' % dbPath, echo=True)

    metadata = MetaData(engine)
    moz_bookmarks = Table('moz_bookmarks', metadata, autoload=True)
    mapper(Bookmarks, moz_bookmarks)

    Session = sessionmaker(bind=engine)
    session = Session()
    return session

if __name__ == "__main__":
    session = loadSession()
    res = session.query(Bookmarks).all()
    res[1].title

В этом фрагменте мы импортируем несколько удобных классов и утилит из sqlalchemy, которые позволяют нам определять такие штуковины, как механизм (тип connection/interface к базе данных), метаданные (каталог таблиц) и сеанс ("дескриптор" базы данных, который позволяет нам запрашивать её). Обратите внимание, что у нас есть файл "places.sqlite". Эта база данных взята из Mozilla Firefox. Если он у вас установлен, поищите его, так как он является отличным испытательным стендом для такого исследования. На моем компьютере с Windows XP он находится в следующем расположении: "C:\Documents and Settings\Mike\Application Data\Mozilla\Firefox\Profiles\f7csnzvk.default". Просто скопируйте этот файл в папку, которую вы будете использовать для сценариев в этой статье. Если вы попытаетесь использовать его на месте и у вас открыт Firefox, у вас могут возникнуть проблемы, так как ваш код может прервать работу Firefox или наоборот.

В вызове create_engine мы устанавливаем echo в значение True. Это приводит к тому, что SQLAlchemy отправляет весь SQL, который он генерирует, в stdout, что весьма удобно для целей отладки. Я бы рекомендовал установить для него значение False, когда вы запускаете свой код в production. Для наших целей наиболее подходяще, эта строка, выглядит следующим образом:

moz_bookmarks = Table('moz_bookmarks', metadata, autoload=True)

Это инструкция SQLAlchemy о попытке автоматической загрузки таблицы "moz_bookmarks". Если это правильная база данных с определённым первичным ключом, это будет отлично работать. Если в таблице нет первичного ключа, надо сделать следущее, чтобы хакнуть ее:

from sqlalchemy import Column, Integer
moz_bookmarks = Table('moz_bookmarks', metadata, 
                      Column("id", Integer, primary_key=True),
                      autoload=True)

Это добавляет дополнительный столбец под названием "id". По сути, это monkey-patching базы данных. Майк Дрискол успешно использовал этот метод, когда приходилось использовать плохо спроектированные базы данных от поставщиков. SQLAlchemy также автоматически увеличит его, если база данных его поддерживает. Средство сопоставления (mapper) сопоставит объект таблицы с классом Bookmarks. Затем создадим сеанс, привязанный к нашему движку, чтобы мы могли делать запросы. res = session.query(Bookmarks).all() в основном означает *SELECT FROM moz_bookmarks** и вернет результат в виде списка объектов Bookmark .

Иногда случается, что в плохой базе данных будет поле, которое, очевидно, должно быть первичным ключом. Если это так, вы можете просто использовать это имя вместо "id". В других случаях вы можете создать уникальный ключ, установив первичный ключ в два столбца (т.е. поля). Вы делаете это, создавая два столбца и устанавливая для "primary_key" значение True для обоих из них. Это намного лучше, чем определять целую таблицу, если в таблице несколько десятков полей.

Давайте перейдем к декларативному методу автозагрузки:

from sqlalchemy import create_engine
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker

engine = create_engine('sqlite:///places.sqlite', echo=True)
Base = declarative_base(engine)

class Bookmarks(Base):
    """"""
    __tablename__ = 'moz_bookmarks'
    __table_args__ = {'autoload':True}

#----------------------------------------------------------------------
def loadSession():
    """"""
    metadata = Base.metadata
    Session = sessionmaker(bind=engine)
    session = Session()
    return session

if __name__ == "__main__":
    session = loadSession()
    res = session.query(Bookmarks).all()
    print res[1].title

Декларативный метод выглядит немного по-другому, да? Попробуем разберём, что тут нового. Во-первых, у нас есть новый импорт: from sqlalchemy.ext.declarative import declarative_base. Здесь используется "declarative_base" для создания класса с использованием нашего движка, из которого затем создаем подкласс Bookmarks. Чтобы указать имя таблицы, мы используем магический метод __tablename__ и, чтобы указать его на автозагрузку, мы создаем словарь __table_args__. Затем в функции loadSession мы получаем объект метаданных следующим образом: metadata = Base.metadata. Остальная часть кода такая же. Декларативный синтаксис обычно упрощает код, поскольку он помещает все это в один класс, а не создает класс и объект таблицы отдельно. Если вам нравится декларативный стиль, вы также можете взглянуть на Elixir, расширение SQLAlchemy, которое было декларативным до того, как они добавили этот стиль в основной пакет. Хотя, может быть не стоит, поскольку (Last Released: Nov 16, 2009).

Явное определение Баз Данных

Бывают случаи, когда нет возможности использовать autoload или просто надо иметь полный контроль над определением таблицы. SQLAlchemy позволяет сделать это почти так же легко. Сначала давайте рассмотрим стандартную (longhand) версию, а затем рассмотрим декларативный стиль.

from sqlalchemy import create_engine, Column, MetaData, Table
from sqlalchemy import Integer, String, Text
from sqlalchemy.orm import mapper, sessionmaker

class Bookmarks(object):
    pass

def loadSession():
    """"""
    dbPath = 'places.sqlite'

    engine = create_engine('sqlite:///%s' % dbPath, echo=True)

    metadata = MetaData(engine)    
    moz_bookmarks = Table('moz_bookmarks', metadata, 
                          Column('id', Integer, primary_key=True),
                          Column('type', Integer),
                          Column('fk', Integer),
                          Column('parent', Integer),
                          Column('position', Integer),
                          Column('title', String),
                          Column('keyword_id', Integer),
                          Column('folder_type', Text),
                          Column('dateAdded', Integer),
                          Column('lastModified', Integer)
                          )

    mapper(Bookmarks, moz_bookmarks)

    Session = sessionmaker(bind=engine)
    session = Session()

if __name__ == "__main__":
    session = loadSession()
    res = session.query(Bookmarks).all()
    print res[1].title

Этот код на самом деле не так уж сильно отличается от кода, который был выше. Часть, которая нас волнует больше всего, — это определение таблицы. Здесь мы используем ключевое слово "Column" для определения каждого столбца. Класс столбца принимает имя столбца, тип, следует ли устанавливать его в качестве первичного ключа и является ли столбец обнуляемым. Он может принять несколько других аргументов, но это те, которые вы увидите чаще всего. В этом примере не показан бит с нулевым значением, потому что я не мог понять, были ли в таблице Firefox эти ограничения.

В любом случае, как только вы определили столбцы, остальная часть кода будет такой же, как и раньше. Тогда давайте перейдем к декларативному стилю:

from sqlalchemy import Column, Integer, String, create_engine
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker

engine = create_engine('sqlite:///places.sqlite', echo=True)
Base = declarative_base(engine)

class Places(Base):
    """"""
    __tablename__ = 'moz_places'

    id = Column(Integer, primary_key=True)
    url = Column(String)
    title = Column(String)
    rev_host = Column(String)
    visit_count = Column(Integer)
    hidden = Column(Integer)
    typed = Column(Integer)
    favicon_id = Column(Integer)
    frecency = Column(Integer)
    last_visit_date = Column(Integer)

    def __init__(self, id, url, title, rev_host, visit_count,
                 hidden, typed, favicon_id, frecency, last_visit_date):
        """"""
        self.id = id
        self.url = url
        self.title = title
        self.rev_host = rev_host
        self.visit_count = visit_count
        self.hidden = hidden
        self.typed = typed
        self.favicon_id = favicon_id
        self.frecency = frecency
        self.last_visit_date = last_visit_date

    def __repr__(self):
        """"""
        return "<places -="" '%s':="" '%s'="">" % (self.id, self.title,
                                                 self.url)

def loadSession():
    """"""
    metadata = Base.metadata
    Session = sessionmaker(bind=engine)
    session = Session()
    return session

if __name__ == "__main__":
    session = loadSession()
    res = session.query(Places).all()
    print res[1].title

Мы немного изменили ситуацию в этом примере, настроив его для другой таблицы: "moz_places". Вы заметите, что вы настроили столбцы здесь, создав переменные класса. Если вы хотите получить доступ к переменным экземпляра, вам необходимо переопределить их в __init__. Если вы этого не сделаете, у вас возникнут довольно запутанные проблемы. __repr__ существует как метод "pretty print". Когда вы печатаете один из объектов "Places", вы получите все, что возвращает __repr__. В остальном код очень похож на другие части, которые мы видели.

Заключение Майка

Как видите, использовать SQLAlchemy для подключения к базам данных очень просто. В большинстве случаев вы заранее узнаете, как устроена база данных, либо из документации, потому что вы создали ее самостоятельно, либо потому, что у вас есть какая-то утилита, которая может вам подсказать. Наличие этой информации полезно, когда дело доходит до определения таблиц в SQLAlchemy, но теперь вы знаете, что попробовать подключаться, можно даже если вы заранее не знаете конфигурацию таблицы! Вот такая веселуха!

Подключение к (mdb) TMGuides

Нужен sqlalchemy-access

import urllib
from sqlalchemy import create_engine, MetaData, Table
from sqlalchemy.orm import mapper, sessionmaker

class Bookmarks(object):
    pass

def loadSessionTmguides():
    connection_string_tmguides = (
        r"DRIVER={Microsoft Access Driver (*.mdb, *.accdb)};"
        r"DBQ=c:\...какой то путь...\DATA\PKM\tmguidesV74.mdb;"
        r"ExtendedAnsiSQL=1;"
    )
    connection_uri = f"access+pyodbc:///?odbc_connect={urllib.parse.quote_plus(connection_string_tmguides)}"

    engine = create_engine(connection_uri)

    metadata = MetaData(engine)
    ngroups = Table('TNGroups', metadata, autoload=True)
    mapper(Bookmarks, ngroups)

    Session = sessionmaker(bind=engine)
    session = Session()
    return session

if __name__ == "__main__":
    session = loadSessionTmguides()
    res = session.query(Bookmarks).all()


Подключение к (mssql) NGuidesSQLNew.mdf

Потребуется SQL Server Native Client 11.0 . В примере используется подключение к штатной базе К3-Мебель 8-й версии с использованием (localdb)\v11.0

from sqlalchemy import create_engine, MetaData, Table
from sqlalchemy.orm import mapper, sessionmaker
from sqlalchemy.engine import URL
 
 
class Bookmarks(object):
    pass
 
def loadSessionLocalNGuidesSQLNew():
 
    connection_string = r"""DRIVER={SQL Server Native Client 11.0};
    SERVER=(localdb)\v11.0;
    DATABASE=C:\ARL8_develop\DataApp\Base\NGuidesSQLNew.mdf;UID=;PWD="""
    connection_url = URL.create("mssql+pyodbc", query={"odbc_connect": connection_string})
 
    engine = create_engine(connection_url)
 
    metadata = MetaData(engine)
    ntypes = Table('NTypes', metadata, autoload=True)
    mapper(Bookmarks, ntypes)
 
    Session = sessionmaker(bind=engine)
    session = Session()
    return session
 
 
if __name__ == "__main__":
    session = loadSessionLocalNGuidesSQLNew()
    res = session.query(Bookmarks).all()
    for a in res:
        print(dict(filter(lambda item: item[0][0] != '_', a.__dict__.items())))

Декларативный вариант

from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker
from sqlalchemy.engine import URL
from sqlalchemy.ext.declarative import declarative_base
 
connection_string = r"""DRIVER={SQL Server Native Client 11.0};
SERVER=(localdb)\v11.0;
DATABASE=C:\ARL8_develop\DataApp\Base\NGuidesSQLNew.mdf;UID=;PWD="""
connection_url = URL.create("mssql+pyodbc", query={"odbc_connect": connection_string})
 
engine = create_engine(connection_url, echo=True)
Base = declarative_base(engine)
 
 
class NTypes(Base):
    """"""
    __tablename__ = 'NTypes'
    __table_args__ = {'autoload':True}
 
#----------------------------------------------------------------------
def loadSessionLocalNGuidesSQLNew():
    """"""
    metadata = Base.metadata
    Session = sessionmaker(bind=engine)
    session = Session()
    return session
 
if __name__ == "__main__":
    session = loadSessionLocalNGuidesSQLNew()
    res = session.query(NTypes).all()
    for a in res:
        print(dict(filter(lambda item: item[0][0] != '_', a.__dict__.items())))