SQLAlchemy — примеры запросов ОРМ

Примеры разных запросов к базе данных через ОРМ SQLAlchemy в python.

Получение данных в SQLAlchemy

first()

Получаем одно значение с помощью first()

save = db.session.query(User).filter_by(user_id=user_id, name=name).first()

Если нам нужно получить не tuple, то добавляем [0]

save = db.session.query(User).filter_by(user_id=user_id, name=name).first()[0]

В этом случае, если не ловить это с помощью исключений try и catch, на запрос [0], будет выкидываться ошибка, если данные не найдены.

Можно этого избежать использую scalar()

scalar()

scalar получает «чистые» данные вида:

2020-10-29 21:31:15.934802+03:00

В то время как first получил бы в этом случае:

(datetime.datetime(2020, 10, 29, 21, 31, 15, 934802, tzinfo=psycopg2.tz.FixedOffsetTimezone(offset=180, name=None)),)

Scalar получит первый элемент запроса или None, в случае если его нет. Если данных несколько вызовет MultipleResultsFound.

brent_oil_current = db.session.query(Base.price).filter_by(name=self.ticker).scalar()

При множественном совпадении, можно так:

brent_oil_current = db.session.query(Base.price).filter_by(name=self.ticker).order_by(db.desc(Base.date)).limit(1).scalar()

IN в SQLAlchemy

ids = [1,2,3]
names = db.session.query(Market.name).filter(MarketShort.id.in_(ids)).all()

Получение данных из моделей SQLAlchemy в виде словарей в списке

Способов получения несколько, что пришлось их выделить в отдельную статью — как вернуть данные в SQLAlchemy в виде списка словарей.

all()

Получение всех данных с помощью .all()

response = db.session.query(UserSetting).filter_by(user_id=user_id).order_by(asc(UserSetting.name)).all()

db.func.max

Получим максимальное число в таблице Setting колонки count:

yesterday = db.session.query(db.func.max(Setting.count)).first()[0]

Сортировки

Получим последние 10 значений из колонки price таблицы Yahoo колонки date:

last_prices = db.session.query(Yahoo.price).order_by(db.desc(Yahoo.date)).limit(10).all()

Или так:

last_prices = db.session.query(Yahoo.price).order_by(Yahoo.date.desc()).limit(10).all()

Получим последнее (отсортированное по дате) значение sp500 по названию ticker:

sp500 = round(
db.session.query(Yahoo.price).filter_by(ticker=self.ticker).order_by(
db.desc(Yahoo.date)).first()[0])

count() и сравнение

count = db.session.query(Yahoo.price).filter(3400>Yahoo.price).order_by(db.desc(Yahoo.date)).limit(20).count()

filter(3216 >Yahoo.price) — здесь сравниваем, больше ли значение в ценах таблицы Yahoo

Альтернативно можно посчитать и так, в том случае, чтобы не делать несколько запросов на выборку всего и подсчета:

count = [x for x in db.session.query(categories_done).filter_by(cat_id=x['id']).all()]
x['count'] = len(count)

Выборка под дате

Например, выборка названий по дате за последние 7 дней:

def get_date_more_then_days(days):
return datetime.today() - timedelta(days=days)

date = get_date_before_days(7)
response = db.session.query(Operation.name).filter_by(user_id=1).filter(
Operation.date > date).order_by(db.asc(Operation.name)).distinct()
shares = [x.name for x in response]

Сохранение данных в SQLAlchemy

Подготавливаем объект для сохранения:

user_setting = UserSetting(user_id=user_id, name=name)

Добавляем данные:

db.session.add(user_setting)

Фиксируем данные и завершаем сессию:

db.session.commit()

Обновление данных в SQLAlchemy

Вариант 1

Обновление данных пользователя, у которого user_id = user_id и name = name. Обновляем у него timer и percent:

db.session.query(User).filter_by(user_id=user_id, name=name).update(
{'timer': int(timer), 'percent': float(percent)}
)

Фиксируем данные и завершаем сессию:

db.session.commit()

Вариант 2

Иногда удобнее использовать другой вариант. Он подойдет, когда нужно проверить, есть ли данные, то обновить их, а если нет, то сохранить новые.

user_setting = db.session.query(UserSetting).filter_by(user_id=user_id, name=name).first()

if user_setting:
user_setting.timer = timer
user_setting.percent = percent
else:
user_setting = UserSetting(user_id=user_id, name=name, timer=timer, percent=percent)
db.session.add(user_setting)
db.session.commit()

Удаление объектов в SQLAlchemy

Удаление не связанных данных

Удаление происходит через метод delete(). У него есть свойство synchronize_session.

Для удаление объекта без обновления состояние объектов в сессии (т.е. если дальше в коде не нужно обращаться к удаленному объекту) надо использовать synchronize_session=False.

db.session.query(categories_done).filter_by(done_id=done_id, cat_id=cat_id).delete(synchronize_session=False)

Удаление связанных данных

Если попытаться удалить связанные данные в SQLAlchemy методом выше, то возникнет ошибка. Для такого удаления подойдет другой способ:

x = db.session.query(Done).filter_by(user_id=user_id, id=id).first()
db.session.delete(x)
db.session.commit()

Получение словарей в SQLAlchemy

Об этом читайте подробнее в статье о vars.

Обработка таблиц вне моделей

query = [serialize_query(x) for x in db.engine.execute(
'select * from table_name order by date_done DESC limit 50'
)]

В этом случае надо также сериализовать данные. Такой пример обработки данных есть здесь.

Как получить выборку по дате

В БД хранится дата в таком виде:

2021-01-16 02:47:54.662908

Мы легко можем получить данные по конкретному дню отправим дату в таком виде:

2021-01-16

Делается это так:

from sqlalchemy import func

date = f'{year}-{day}-{month}'

response = [x.serialize for x in
db.session.query(Done)
.filter_by(user_id=flask_login.current_user.id)
.filter(func.DATE(Done.time) == date)
.order_by(db.desc(Done.time)).all()
]

Получение пользователей по их ролям

О добавление ролей на сайт можно почитать здесь: пользователи и роли на Flask.

Так:

for user in db.session.query(User).filter_by(active=True).all():
for role in user.roles:
if role == 'admin':

Или:

users = db.session.query(User).filter_by(active=True).all()
for x in users:
for role in x.roles:
if role == 'admin':

То же самое, если используется Flask-Security:

for user in db.session.query(User).filter_by(active=True).filter().all():
if user.has_role('user'):

Not None

Фильтрация пользователей. Забирает только тех из них у кого заполнен телеграм-id:

for user in db.session.query(User).filter_by(active=True).filter(User.telegram_id.isnot(None)).all():