Python

• FastApi
• Общие команды
• Маршрутизация
• Pydantic
• Jinja2
• Авторизация и аутентификация
• SQLAlchemy
• Sqlalchemy
• ORM режим
• Core режим
• Пример проекта
• Alembic
• Установка и настройка
• Миграции
• Модули
• Описание модулей
• jsonschema
• Pydantic 2
• Rabbitmq
• Авторизация через ВК
• Тестирование Playwright
• Начало
• Локаторы
• Actions
• События (Events)
• Аутентификация
• Pytest & Playwright
• Дополнительные возможности
• Ожидание

FastApi

Общие команды

Установка

pip install fastapi uvicorn

Ручной запуск (api - имя файла, app - имя объекта FastApi)

uvicorn api:app --port 8000 --reload

Запуск uvicorn из python скрипта

Файл main.py

from uvicorn import run
...
app = FastAPI()
...
if __name__ == '__main__':
    run(app="main:app", host='0.0.0.0', port=8000, workers=4, log_level='warning')
    #run(app="main:app", host='0.0.0.0', port=8000, reload=True)

curl запросы

curl -X 'GET' 'http://127.0.0.1:8000/todo' -H 'accept: application/json' 

curl -X 'POST' \ 
'http://127.0.0.1:8000/todo' \ 
-H 'accept: application/json' \ 
-H 'Content-Type: application/json' \ 
-d '{ 
"id": 1, 
"item": "First Todo is to finish this book!" 
}'

Автоматическая документация

Swagger

http://ip:port/docs

Redoc

http://ip:port/redoc

Шаблоны Jinja

Поддерживает шаблоны Jinja при выводе данных (вплоть до циклов).

Маршрутизация

Добавление маршрутов

Основной файл:

from fastapi import FastAPI 
from todo import todo_router 
 
app = FastAPI() 

@app.get("/") 
async def welcome() -> dict: 
    return { 
        "message": "Hello World" 
        } 

app.include_router(todo_router)

Файл дополнительных маршрутов

from fastapi import APIRouter 
 
todo_router = APIRouter()

todo_list = [] 
 
@todo_router.post("/todo") 
async def add_todo(todo: dict) -> dict: 
    todo_list.append(todo) 
    return {"message": "Todo added successfully"} 

@todo_router.get("/todo") 
async def retrieve_todos() -> dict: 
    return {"todos": todo_list}

Автоматическое добавление маршрутов в основной файл app из файлов в директории data/plugins имеющих шаблон имени объекта APIRouter modulename_router 

fpath = os.path.join('data', 'plugins')
flist = os.listdir(fpath)
sys.path.insert(0, fpath)
for fname in flist:
    if fname not in ['__pycache__', '__init__.py']:
        m = os.path.splitext(fname)[0]
        impmod = importlib.import_module(m)
        router_name = f'{m}_router'
        if router_name in dir(impmod):
            router_mod = getattr(impmod, router_name)
            app.include_router(router_mod)

Получаемые параметры

Параметры пути:

from fastapi import Path

@todo_router.get("/todo/{todo_id}") 
async def get_single_todo(todo_id: int = Path(..., title="The ID of the todo to retrieve.")) -> dict: 
    for todo in todo_list: 
        if todo.id == todo_id: 
            return {"todo": todo} 
    return { 
        "message": "Todo with supplied ID doesn't exist." 
        }

В Path ... - параметр пути обязательный, None - не обязательный

Параметры запроса (после ? в запросе):

async query_route(query: str = Query(None): 
    return query

Передаваемые параметры

При помощи response_model можно фильтровать отдаваемые данные. Т е можно в отдаваемой модели указать неполный набор.

from typing import List 
 
class TodoItem(BaseModel): 
    item: str 
 
class TodoItems(BaseModel): 
    todos: List[TodoItem]

@todo_router.get("/todo", response_model=TodoItems) 
async def retrieve_todo() -> dict: 
    return { 
        "todos": todo_list 
        }

У содержащихся в списке словарей будет оставлен только item

Исключения

Класс HTTPException принимает три аргумента: 

• status_code: Код состояния, который будет возвращен для этого сбоя
• detail: Сопроводительное сообщение для отправки клиенту 
• headers: Необязательный параметр для ответов, требующих заголовков

from fastapi import APIRouter, Path, HTTPException, status 

@todo_router.get("/todo/{todo_id}") 
async def get_single_todo(todo_id: int = Path(..., title="The ID of the todo to retrieve.")) -> dict: 
    for todo in todo_list:
      if todo.id == todo_id: 
        return { "todo": todo } 
    raise HTTPException( 
        status_code=status.HTTP_404_NOT_FOUND, 
        detail="Todo with supplied ID doesn't exist", 
        )

Можно переопределить код успешного возврата в декораторе

@todo_router.post("/todo", status_code=201) 
async def add_todo(todo: Todo) -> dict: 
  todo_list.append(todo) 
  return { "message": "Todo added successfully." }

Pydantic

class Item(BaseModel): 
    item: str 
    status: str 
 
class Todo(BaseModel):
    id: int 
    item: Item

Jinja2

{{ variable | filter_name(*args) }}

{{ todo.item | default('This is a default todo item') }}

{{ 3.142 | int }} 
3 
{{ 31 | float }} 
31.0

{{ ['Packt', 'produces', 'great', 'books!'] | join(' ') }} 
Packt produces great books!

Todo count: {{ todos | length }} 
Todo count: 4

{% if user %}
    Hello, {{ user.name }}!
{% else %}
    Hello, Unknown!
{% endif %}

{% for comment in comments %}
   <b>{{ comment }}</b>
{% endfor %}

Авторизация и аутентификация

Ссылки:

Fastapi users документация

Role-based authentification

Подготовка проекта

python -m venv --system-site-packages env
python -m pip install fastapi uvicorn

SQLAlchemy

Sqlalchemy

pip install sqlalchemy

pip install psycopg2

pip install psycopg2-binary

pip  install  pymysql

from sqlalchemy import create_engine
engine = create_engine('postgresql+psycopg2://username:password@localhost:5432/mydb')
engine = create_engine('mysql+pymysql://cookiem:chip@mysql01.com/cookies', pool_recycle=3600)
engine = create_engine('sqlite:///cookies.db')
engine2 = create_engine('sqlite:///:memory:')

Создание engine не создает фактического соединения с БД. 

connection = engine.connect()

Сырой запрос:

result = connection.execute("select * from orders").fetchall()

from sqlalchemy import MetaData
convention = {
'all_column_names': lambda constraint, table: '_'.join([
column.name for column in constraint.columns.values()
]),
'ix': 'ix__%(table_name)s__%(all_column_names)s',
'uq': 'uq__%(table_name)s__%(all_column_names)s',
'ck': 'ck__%(table_name)s__%(all_column_names)s',
'fk': ('fk__%(table_name)s__%(all_column_names)s' '%(referred_table_name)s'),
'pk': 'pk__%(table_name)s'
}
metadata_obj = MetaData(naming_convention=convention)

from sqlalchemy import ForeignKeyConstraint
album.append_constraint(ForeignKeyConstraint(['ArtistId'], ['artist.ArtistId']))

engine = create_engine(...)
metadata = MetaData()
cookie_tbl = Table('cookies', metadata, autoload_with=engine)
s = select(cookie_tbl)
with engine.connect() as conn:
    m = conn.execute(s)
    print(m.keys())

from sqlalchemy import create_engine, MetaData

engine = create_engine(...)
metadata = MetaData()
metadata.reflect(bind=engine)
for table in metadata.sorted_tables:
    print(table.name)
#Потом получить объект таблицы: 
mytable = metadata.tables['mytable']

from sqlalchemy.ext.automap import automap_base
from sqlalchemy import create_engine

Base = automap_base()
engine = create_engine('sqlite:///Chinook_Sqlite.sqlite')
Base.prepare(engine, reflect=True)
# данные о классах загружены. 
#Например для получения списка классов:
Base.classes.keys()
Artist = Base.classes.Artist # создание классов
#Внешние связи - в свойстве <related_object>_collection
artist = session.query(Artist).first()
for album in artist.album_collection:
    print('{} - {}'.format(artist.Name, album.Title))

ins = cookies.insert().values(
    cookie_name="chocolate chip",
    cookie_recipe_url="http://some.aweso.me/cookie/recipe.html",
    cookie_sku="CC01",
    quantity="12",
    unit_cost="0.50"
    )
    result = connection.execute(ins)
    print(result.inserted_primary_key)

from sqlalchemy import insert
ins = insert(cookies).values( 
    cookie_name="chocolate chip",
    cookie_recipe_url="http://some.aweso.me/cookie/recipe.html",
    cookie_sku="CC01",
    quantity="12",
    unit_cost="0.50"
)

ins = cookies.insert()
result = connection.execute(
    ins, 
    cookie_name='dark chocolate chip', 
    cookie_recipe_url='http://some.aweso.me/cookie/recipe_dark.html',
    cookie_sku='CC02',
    quantity='1',
    unit_cost='0.75'
)
result.inserted_primary_key

inventory_list = [ 
    {
        'cookie_name': 'peanut butter',
        'cookie_recipe_url': 'http://some.aweso.me/cookie/peanut.html',
        'cookie_sku': 'PB01',
        'quantity': '24',
        'unit_cost': '0.25'
    },
    {
        'cookie_name': 'oatmeal raisin',
        'cookie_recipe_url': 'http://some.okay.me/cookie/raisin.html',
        'cookie_sku': 'EWW01',
        'quantity': '100',
        'unit_cost': '1.00'
    }
]
result = connection.execute(ins, inventory_list)

from sqlalchemy.sql import select
s = select([cookies]) 
rp = connection.execute(s)
results = rp.fetchall()

s = cookies.select()
rp = connection.execute(s)
results = rp.fetchall()

s = cookies.select()
rp = connection.execute(s) 
for record in rp:
    print(record.cookie_name)

s = select([cookies.c.cookie_name, cookies.c.quantity])

s = select([cookies.c.cookie_name, cookies.c.quantity])

s = s.order_by(cookies.c.quantity)

s = s.order_by(desc(cookies.c.quantity))

rp = connection.execute(s)

s = select([cookies.c.cookie_name, cookies.c.quantity])
s = s.order_by(cookies.c.quantity)
s = s.limit(2)
rp = connection.execute(s)

from sqlalchemy.sql import func
s = select([func.sum(cookies.c.quantity)])
rp = connection.execute(s)
print(rp.scalar())

s = select([func.count(cookies.c.cookie_name)])
rp = connection.execute(s)
record = rp.first()
print(record.keys()) # ключи могут быть разные.
print(record.count_1) 

s = select([func.count(cookies.c.cookie_name).label('inventory_count')]) 
rp = connection.execute(s)
record = rp.first()
print(record.keys())
print(record.inventory_count)

s = select([cookies]).where(cookies.c.cookie_name == 'chocolate chip')
rp = connection.execute(s)

• between(cleft, cright) Значение столбца между cleft и cright
• concat(column_two) Объединение column и column_two
• distinct() Находит только уникальные значения в столбце
• in_([list]) Только если значения столбца в списке
• is_(None) Проверка на пустые значения 
• contains(string) Значение столбца содержит строку
• endswith(string) Оканчивается строкой, зависит от больших букв
• like(string) зависит от больших букв
• startswith(string) зависит от больших букв
• ilike(string) не зависит от больших букв 
• Есть отрицательные модификаторы not<method>, исключение - метод isnot() 

s = select([cookies]).where(cookies.c.cookie_name.like('%chocolate%'))
rp = connection.execute(s)
for record in rp.fetchall():
    print(record.cookie_name)

s = select([cookies.c.cookie_name, 'SKU-' + cookies.c.cookie_sku]) - добавит строку 'SKU-'

from sqlalchemy import cast 
s = select([cookies.c.cookie_name,
          cast((cookies.c.quantity * cookies.c.unit_cost),
               Numeric(12,2)).label('inv_cost')]) 
for row in connection.execute(s):
    print('{} - {}'.format(row.cookie_name, row.inv_cost))

from sqlalchemy import and_, or_, not_
s = select([cookies]).where(
    and_(
        cookies.c.quantity > 23,
        cookies.c.unit_cost < 0.40
   )
)

first_row = results[0] 
first_row[1] 
first_row.cookie_name 
first_row[cookies.c.cookie_name]

• first() - Первая запись и закрытие соединения
• fetchone() - Одна запись и оставляет открытый курсор для последующих запросов
• scalar() - Одно значение если запрос возвращает одно значение в одной строке
• rp.keys() - список столбцов

from sqlalchemy import update
u = update(cookies).where(cookies.c.cookie_name == "chocolate chip")
u = u.values(quantity=(cookies.c.quantity + 120)) 
result = connection.execute(u)

from sqlalchemy import delete
u = delete(cookies).where(cookies.c.cookie_name == "dark chocolate chip")
result = connection.execute(u)

columns = [orders.c.order_id, users.c.username, users.c.phone,
           cookies.c.cookie_name, line_items.c.quantity,
           line_items.c.extended_cost]
cookiemon_orders = select(columns)
cookiemon_orders = cookiemon_orders.select_from(orders.join(users).join( 
                        line_items).join(cookies)).where(users.c.username ==
                            'cookiemon')
result = connection.execute(cookiemon_orders).fetchall()

columns = [users.c.username, func.count(orders.c.order_id)]
all_orders = select(columns)
all_orders = all_orders.select_from(users.outerjoin(orders)) 
all_orders = all_orders.group_by(users.c.username)
result = connection.execute(all_orders).fetchall()

columns = [users.c.username, func.count(orders.c.order_id)] 
all_orders = select(columns)
all_orders = all_orders.select_from(users.outerjoin(orders))
all_orders = all_orders.group_by(users.c.username) 
result = connection.execute(all_orders).fetchall()

def get_orders_by_customer(cust_name, shipped=None, details=False):
    columns = [orders.c.order_id, users.c.username, users.c.phone]
    joins = users.join(orders)
    if details:
        columns.extend([cookies.c.cookie_name, line_items.c.quantity,
                       line_items.c.extended_cost])
        joins = joins.join(line_items).join(cookies)
    cust_orders = select(columns)
    cust_orders = cust_orders.select_from(joins)


    cust_orders = cust_orders.where(users.c.username == cust_name)
    if shipped is not None:
        cust_orders = cust_orders.where(orders.c.shipped == shipped)
    result = connection.execute(cust_orders).fetchall()
    return result
get_orders_by_customer('cakeeater') 
get_orders_by_customer('cakeeater', details=True) 
get_orders_by_customer('cakeeater', shipped=True) 
get_orders_by_customer('cakeeater', shipped=False) 
get_orders_by_customer('cakeeater', shipped=False, details=True) 

ORM режим

Таблица это класс с требованиями:

• Потомок объекта, возвращаемого функцией declarative_base
• Включает __tablename__ с именем таблицы
• Включает 1+ атрибутов, являющихся объектом Column
• При определении не включает имя столбца в конструкторе Column, имя столбца = имя атрибута
• 1+ атрибутов определяют первичный ключ
• __table_args__ свойства таблицы (ограничения,...)

from sqlalchemy import Table, Column, Integer, String
from sqlalchemy.ext.declarative import declarative_base
			
	Base = declarative_base() 
	
	class Cookie(Base): 
		__tablename__ = 'cookies' 
		__table_args__ = (CheckConstraint('quantity >= 0', name='quantity_positive'),)
		cookie_id = Column(Integer(), primary_key=True) 
		cookie_name = Column(String(50), index=True)
		quantity = Column(Integer())

Создание таблиц

from sqlalchemy import create_engine
from dataclasses import Base

engine = create_engine(...)
Base.metadata.create_all(engine)

Ограничения

__table_args__ = (ForeignKeyConstraint(['id'], ['other_table.id']), CheckConstraint(unit_cost >= 0.00', name='unit_cost_positive'))

Внешние связи

• Определяется столбец с ForeignKey
• Определяется дополнительный атрибут с relationship и необязательным backref
• При определении backref, relationship будет определен в указанном классе с указанным именем.

Один к одному:

cookie = relationship("Cookie", uselist=False)

Один ко многим:

user = relationship("User", backref=backref('orders'))

На себя (дерево) - неоднозначное решение. 

class Employee(Base):
	__tablename__ = 'employees'
	id = Column(Integer(), primary_key=True)
	manager_id = Column(Integer(), ForeignKey('employees.id'))
	name = Column(String(255), nullable=False)
	manager = relationship("Employee", backref=backref('reports'), remote_side=[id])

Сессии

В ORM режиме, сессия упаковывает

• соединение с БД через engine и предоставляет словарь объектов, загруженных через сессию или ассоциированных с сессией
• транзакции, которые открыты до коммита сессии

Это похожая на хэш-систему, состоящую из списка объектов, таблиц и ключей. Сессия создается через sessionmaker, один раз. Соединяется с базой в момент необходимости

from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker 

engine = create_engine('sqlite:///:memory:')
Session = sessionmaker(bind=engine) 
session = Session()

Состояния объекта в сессии:

Просмотр состояния:

from sqlalchemy import inspect
insp = inspect(cc_cookie)

Отключение объекта от сессии:

session.expunge(cc_cookie)

Просмотр списка атрибутов и выяснение, что было модифицировано

for attr, attr_state in insp.attrs.items():
	if attr_state.history.has_changes(): 
		print('{}: {}'.format(attr, attr_state.value))
		print('History: {}\n'.format(attr_state.history))

Добавление данных

Создаем объект класса, добавляем в сессию и коммитим. Множественное добавление данных:

dcc = Cookie(...)
mcc = Cookie(...)
session.add(dcc)
session.add(mcc)
session.flush()

flush: не выполняет коммит и не завершает транзакцию, но получает id. Потом нужно в пределах сессии сделать commit. Commit в пределах чужой сессии не влияет. Дальше можно использовать объект.

Если дальше не нужно выполнять операции с объектами:

session.bulk_save_objects([dcc,mcc])
session.commit()

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

cookies = session.query(Cookie).all() 
print(cookies)

Если использовать через итератор, то без all:

for cookie in session.query(Cookie): 
    print(cookie)

Получение определенных столбцов:

cookies = session.query(Cookie.cookie_id).all()

Сортировка:

session.query(Cookie).order_by(Cookie.quantity).all()
.order_by(desc(Cookie.quantity))

Ограничения через срезы или .limit(2)

Встроенные функции:

from sqlalchemy import func
inv_count = session.query(func.sum(Cookie.quantity)).scalar() 
print(inv_count)
rec_count = session.query(func.count(Cookie.cookie_name)).first()

Метки: можно добавить для дальнейшего обращения

rec_count = session.query(func.count(Cookie.cookie_name).label('inventory_count')).first()
print(rec_count.inventory_count)

Фильтрация

record = session.query(Cookie).filter(cookie_name=='chocolate chip').first() 
record = session.query(Cookie).filter_by(cookie_name='chocolate chip').first()
query = session.query(Cookie).filter(Cookie.cookie_name.like('%chocolate%'))
query = session.query(Cookie).filter(Cookie.quantity > 23, Cookie.unit_cost < 0.40)

Изменение строк при выводе

results = session.query(Cookie.cookie_name, 'SKU-' + Cookie.cookie_sku).all()
query = session.query(Cookie.cookie_name,
      cast((Cookie.quantity * Cookie.unit_cost),
           Numeric(12,2)).label('inv_cost'))

Join:

query = session.query(Order.order_id, User.username)
results = query.join(User).all()

query = session.query(User.username, func.count(Order.order_id))
query = query.outerjoin(Order).group_by(User.username)

Group:

query = session.query(User.username, func.count(Order.order_id))
query = query.outerjoin(Order).group_by(User.username)

Сырые запросы

from sqlalchemy import text
query = session.query(User).filter(text("username='cookiemon'"))

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

Через объект

query = session.query(Cookie) 
cc_cookie = query.filter(Cookie.cookie_name == "chocolate chip").first() 
cc_cookie.quantity = cc_cookie.quantity + 120
session.commit()

Через метод update

query = session.query(Cookie)
query = query.filter(Cookie.cookie_name == "chocolate chip")
query.update({Cookie.quantity: Cookie.quantity - 20})

Удаление

session.delete(dcc_cookie)
session.commit()

Исключения

from sqlalchemy.orm.exc import MultipleResultsFound 
try:
	results = session.query(Cookie).one()
except MultipleResultsFound as error: 
	print('We found too many cookies... is that even possible?')

Транзакции

В ORM транзакция создается автоматически до очередного коммита.

session.add(order)
try:
	session.commit()
except IntegrityError as error: 
	session.rollback() 

Пример структурирования

db.py

from datetime import datetime
from sqlalchemy import (MetaData, Table, Column, Integer, Numeric, String,
                        DateTime, ForeignKey, Boolean, create_engine)
class DataAccessLayer:
    connection = None
    engine = None
    conn_string = None
    metadata = MetaData()
    cookies = Table('cookies',
                    metadata,
                    Column('cookie_id', Integer(), primary_key=True),
                    Column('cookie_name', String(50), index=True),
                    Column('cookie_recipe_url', String(255)),
                    Column('cookie_sku', String(55)),
                    Column('quantity', Integer()),
                    Column('unit_cost', Numeric(12, 2))
              ) 
			  
    def db_init(self, conn_string): 
        self.engine = create_engine(conn_string or self.conn_string)
        self.metadata.create_all(self.engine)
        self.connection = self.engine.connect()
dal = DataAccessLayer()

app.py

from db import dal 
from sqlalchemy.sql import select

def get_orders_by_customer(cust_name, shipped=None, details=False):
    columns = [dal.orders.c.order_id, dal.users.c.username, dal.users.c.phone]
    joins = dal.users.join(dal.orders) 
    if details:
        columns.extend([dal.cookies.c.cookie_name,
                        dal.line_items.c.quantity,
                        dal.line_items.c.extended_cost])
        joins = joins.join(dal.line_items).join(dal.cookies)
    cust_orders = select(columns)
    cust_orders = cust_orders.select_from(joins).where(
        dal.users.c.username == cust_name)
    if shipped is not None:
        cust_orders = cust_orders.where(dal.orders.c.shipped == shipped)
    return dal.connection.execute(cust_orders).fetchall()

test.py 

import unittest
class TestApp(unittest.TestCase): 
    
	@classmethod
    def setUpClass(cls): 
        dal.db_init('sqlite:///:memory:')
	
	def test_one(self):
		res = get_orders_by_customer('', False)
		self.assert_equal(res, [])

Core режим

Сначала необходимо определить, как данные хранятся в таблице. Варианты определения:

• Объект Table
• Декларативный класс
• Получение структуры из базы данных

Сопоставление типов

Metadata

Каталог объектов Table с опциональной информацией о engine и соединении.

from sqlalchemy import MetaData
metadata = MetaData()

Создание таблицы

metadata.create_all(engine)

Метод ...create_all не пересоздает таблицы.

Объект таблицы состоит из названия, переменной метаданных и столбцов.

from sqlalchemy import Table, Column, Integer, Numeric, String, ForeignKey
from datetime import datetime
from sqlalchemy import DateTime

cookies = Table('cookies', metadata,
    Column('cookie_id', Integer(), primary_key=True), 
    Column('cookie_name', String(50), index=True), 
    Column('cookie_recipe_url', String(255)),
    Column('cookie_sku', String(55)),
    Column('quantity', Integer()),
    Column('unit_cost', Numeric(12, 2)) 
)

users = Table('users', metadata,
    Column('user_id', Integer(), primary_key=True),
    Column('username', String(15), nullable=False, unique=True), 
    Column('email_address', String(255), nullable=False),
    Column('phone', String(20), nullable=False),
    Column('password', String(25), nullable=False),
    Column('created_on', DateTime(), default=datetime.now), 
    Column('updated_on', DateTime(), default=datetime.now, onupdate=datetime.now) 
)

Класс Column

• название столбца
  • тип данных
    • в String обязательно указывать длину
    • Numeric(11,2) означает 11
  • доп. параметры

primary_key=True
index=True
nullable=False
unique=True
default=datetime.now
onupdate=datetime.now

Ключи, ограничения и индексы

Могут быть определены в конструкторе столбца (primary_key=True) или позже в конструкторе таблицы.

from sqlalchemy import PrimaryKeyConstraint, UniqueConstraint, CheckConstraint

users = Table(...
             PrimaryKeyConstraint('user_id', name='user_pk'),
             UniqueConstraint('username', name='uix_username'),
             CheckConstraint('unit_cost >= 0.00', name='unit_cost_positive'),
             ...)

Множественные ключи перечисляются через запятую.

from sqlalchemy import Index
 
Index('ix_cookies_cookie_name', 'cookie_name')
Index('ix_test', mytable.c.cookie_sku, mytable.c.cookie_name)

Внешние связи

from sqlalchemy import ForeignKey
orders = Table('orders', metadata,
    Column('order_id', Integer(), primary_key=True),
    Column('user_id', ForeignKey('users.user_id')),
    Column('shipped', Boolean(), default=False)
)
line_items = Table('line_items', metadata,
    Column('line_items_id', Integer(), primary_key=True),
    Column('order_id', ForeignKey('orders.order_id')),
    Column('cookie_id', ForeignKey('cookies.cookie_id')),
    Column('quantity', Integer()),
    Column('extended_cost', Numeric(12, 2))
)

Связь для поля order_id:

Column('user_id', ForeignKey('users.user_id'))
#При желании - ограничение
ForeignKeyConstraint(['order_id'], ['orders.order_id'])

Добавление данных

from sqlalchemy import insert
перем = таблица.insert().values()

Лучше (?) вариант

from sqlalchemy import insert
перем = insert(таблица).values()

Строковое представление запроса

str(перем)

Компиляция запроса

перем.compile()

перем.compile().params

Примеры

with engine.connect() as connection:
	metadata = ...
	cookies = Table...
	ins = insert(cookies).values(...)
	res = connection.execute(ins)
	#res.inserted_primary_key - какой в будущем будет ключ (сейчас фактически в БД нет данных)
	connection.commit()

ins = cookies.insert()
inventory_list = [ 
	{
		'cookie_name': 'peanut butter',
		'cookie_recipe_url': 'http://some.aweso.me/cookie/peanut.html',
	},
	{
		'cookie_name': 'oatmeal raisin',
		'cookie_recipe_url': 'http://some.okay.me/cookie/raisin.html',
	}
]
result = connection.execute(ins, inventory_list)

from sqlalchemy import create_engine
from sqlalchemy import MetaData
from sqlalchemy import Table, Column, Integer, Numeric, String, ForeignKey

metadata = MetaData()
cookies = Table('cookies', metadata,
	Column('cookie_id', Integer(), primary_key=True), 
	Column('cookie_name', String(50), index=True), 
	Column('cookie_recipe_url', String(255))
)
engine = create_engine('sqlite:///:memory:')
connection = engine.connect()
metadata.create_all(engine)

from sqlalchemy import insert

ins = cookies.insert().values(
	cookie_name="chocolate chip",
	cookie_recipe_url="http://some.aweso.me/cookie/recipe.html"
	)
print(str(ins))

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

from sqlalchemy.sql import select
s = select(cookies) 
rp = connection.execute(s)

Список столбцов

rp.keys()

Получение результата

results = rp.fetchall()

Доступ возможен по:

Сортировка

s = select(cookies.c.cookie_name, cookies.c.quantity)
s = s.order_by(cookies.c.quantity)
s = s.order_by(desc(cookies.c.quantity))

Ограничения количества

s = s.limit(2)

Встроенные функции

from sqlalchemy.sql import func
s = select([func.sum(cookies.c.quantity)])
rp = connection.execute(s)
print(rp.scalar())

Фильтрация

s = select([cookies]).where(cookies.c.cookie_name == 'chocolate chip')

Внутри where можно использовать and_, or_, not_

from sqlalchemy import and_, or_, not_
s = select([cookies]).where(
	and_(
		cookies.c.quantity > 23,
		cookies.c.unit_cost < 0.40
	)
)

Join

columns = [orders.c.order_id, users.c.username, users.c.phone,
		cookies.c.cookie_name, line_items.c.quantity, line_items.c.extended_cost]
cookiemon_orders = select(*columns)
cookiemon_orders = cookiemon_orders.select_from(orders.join(users).join(line_items).join(cookies)).where(users.c.username == 'cookiemon')
result = connection.execute(cookiemon_orders).fetchall()
for row in result:
	print(row)

Для outerjoin: join -> outerjoin

Алиасы

manager = employee_table.alias('mgr')

Grouping:

columns = [users.c.username, func.count(orders.c.order_id)] 
all_orders = select(columns)
all_orders = all_orders.select_from(users.outerjoin(orders))
all_orders = all_orders.group_by(users.c.username)

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

from sqlalchemy import update
u = update(cookies).where(cookies.c.cookie_name == "chocolate chip")
u = u.values(quantity=(cookies.c.quantity + 120)) 
result = connection.execute(u)

Удаление данных:

from sqlalchemy import delete
u = delete(cookies).where(cookies.c.cookie_name == "dark chocolate chip")
result = connection.execute(u)

Сырые запросы (raw)

result = connection.execute("select * from orders").fetchall()

Обработка исключений

AttributeError - ошибка набора данных
IntegrityError - ошибка ограничений
Стандартная обработка подходит если исполняется один независимый запрос.

try:
	result = connection.execute(ins)
except IntegrityError as error: 
	print(error.orig.message, error.params)

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

transaction = connection.begin() 
try:
	...
	transaction.commit() 
except IntegrityError as error:
	transaction.rollback()

Пример:

transaction = connection.begin() 
cookies_to_ship = connection.execute(s).fetchall() 
try:
	for cookie in cookies_to_ship:
		u = update(cookies).where(cookies.c.cookie_id == cookie.cookie_id)
		u = u.values(quantity = cookies.c.quantity-cookie.quantity)
		result = connection.execute(u)
	u = update(orders).where(orders.c.order_id == order_id)
	u = u.values(shipped=True)
	result = connection.execute(u)
	print("Shipped order ID: {}".format(order_id))
	transaction.commit() 
except IntegrityError as error:
	transaction.rollback() 

Пример проекта

Структура проекта

Предварительная настройка

Для работы примера необходимо установить docker. 

Клонировать проекта с репозитория 

git clone https://gitverse.ru/bobrobot/alembictemplate.git

Перейти в директорию проекта, создать виртуальное окружение и активировать 

cd alembictemplate
python3 -m venv env
source env/bin/activate

Установить дополнительные модули 

pip install -r requirements.txt

Перейти в директорию docker и в файле docker-compose.yml настроить пути, имя БД, логин и пароль к новой базе данных. 

services:
  postgres:
    image: postgres:latest
    environment:
      POSTGRES_DB: "learnsqlalchemy"
      POSTGRES_USER: "learner"
      POSTGRES_PASSWORD: "StrongPassword123"
      PGDATA: "/home/sergey/projects/alembictemplate/docker/data/pgdata"
    volumes:
      - .:/docker-entrypoint-initdb.d
      - mydata:/home/sergey/projects/alembictemplate/docker/data
    ports:
      - "5430:5432"
volumes:
  mydata:

В файле docker-entrypoint-initdb.d/main.sql изменить имя БД, логин и пароль.

CREATE DATABASE learnsqlalchemy;
CREATE USER learner WITH PASSWORD 'StrongPassword123';
ALTER ROLE learner WITH PASSWORD 'StrongPassword123';
GRANT ALL PRIVILEGES ON DATABASE learnsqlalchemy to learner;

В директории docker запустить контейнер БД в фоновом режиме. 

docker compose up -d

Для остановки контейнера: 

docker compose stop

Сейчас, запустив контейнер, при помощи консольного клиента psql можно проверить соединение с базой данных для пользователя. 

psql -d learnsqlalchemy -U learner -W -h 127.0.0.1 -p 5430

Для работы с настройками в формате json используется библиотека src/libsettings.py Описание библиотеки Создать папку src, скопировать из проекта библиотеку libsettings.py 

Настройки системы

Файлы основных настроек расположены в conf/settings/  Файл base.py 

'''Loading settings to project'''
import os
from src.libsettings import Jsettings

settingspath = os.path.join('conf', 'settings', 'settings.json')
schemapath = os.path.join('conf', 'settings', 'schema.json')
# dev settings
mysettings = Jsettings(settingsfname= settingspath,
                       schemafname=schemapath)
mysettings.load_settings()

Файл schema.json 

{
    "type": "object",
    "properties": {
        "db_username": {"type": "string"},
        "db_password": {"type": "string"},
        "db_host": {"type": "string"},
        "db_port": {"type": "string"},
        "db_name": {"type": "string"}
    }, 
    "required": ["db_username", "db_password", "db_host",
                "db_port", "db_name"]
}

Файл settings.json 

{
    "db_username": "learner",
    "db_password": "StrongPassword123",
    "db_host": "127.0.0.1",
    "db_port": "5430",
    "db_name": "learnsqlalchemy"
}

Файл base.py проверяет схему и создает объект настроек mysettings из файла settings.json. Для получения объекта настроек нужно импортировать объект mysettings в нужном модуле. В данный момент присутствуют настройки базы данных с префиксом db_*

Если такая усложненная система управления настройками покажется излишней, возможно использовать экспорт настроек напрямую из файла.

Инициализация базы данных

Создаем папку db, в ней создаем файл initializer.py 

'''Db classes and  initialization'''
from sqlalchemy import create_engine
from sqlalchemy.orm import declarative_base
from conf.settings.base import mysettings

#load engine settings
engine = create_engine(
    "postgresql+psycopg2://{db_username}:{db_password}@{db_host}:{db_port}/{db_name}".format(
        db_username=mysettings.db_username,
        db_password=mysettings.db_password,
        db_host=mysettings.db_host,
        db_port=mysettings.db_port,
        db_name=mysettings.db_name
    ),
    echo=True)
Base = declarative_base()

Здесь только создается engine для подключения к БД и класс Base. При настройке структуры таблиц данный файл будет обновлен, сейчас нужен только класс Base.

Настройка системы версионирования базы данных alembic

Инициализируем alembic в корне проекта. 

alembic init alembic

В корне проекта будет создан файл alembic.ini, будет создана папка alembic с файлами инициализации. В большинстве инструкций параметры подключения задаются в файле alembic.ini однако, для доступа к настройкам из единой точки будет изпользоваться способ установки параметров в файле env.py Поэтому в файле alembic.ini переменная sqlalchemy.url должна быть закомментирована. Часть файла alembic.ini 

#sqlalchemy.url = driver://user:pass@localhost/dbname

В файле env.py

• импортируем путь 

import os
import sys

sys.path.append(os.getcwd())

Импортируем настройки, создаем строку соединения и создаем закоментированный ранее в файле alembic.ini параметр sqlalchemy.url

from conf.settings.base import mysettings

connstring = "postgresql+psycopg2://{db_username}:{db_password}@{db_host}:{db_port}/{db_name}".format(
        db_username=mysettings.db_username,
        db_password=mysettings.db_password,
        db_host=mysettings.db_host,
        db_port=mysettings.db_port,
        db_name=mysettings.db_name
    )
config.set_main_option(name="sqlalchemy.url", value=connstring)

Импортируем db.initializer и создаем метаданные 

import db.initializer

target_metadata = db.initializer.Base.metadata

Остальные параметры оставляем неизменными. Результирующий файл настроек окружения alembic env.py: 

from logging.config import fileConfig
import os
import sys

from sqlalchemy import engine_from_config
from sqlalchemy import pool

from alembic import context

from conf.settings.base import mysettings

sys.path.append(os.getcwd())

# this is the Alembic Config object, which provides
# access to the values within the .ini file in use.
config = context.config

# Interpret the config file for Python logging.
# This line sets up loggers basically.
if config.config_file_name is not None:
    fileConfig(config.config_file_name)

connstring = "postgresql+psycopg2://{db_username}:{db_password}@{db_host}:{db_port}/{db_name}".format(
        db_username=mysettings.db_username,
        db_password=mysettings.db_password,
        db_host=mysettings.db_host,
        db_port=mysettings.db_port,
        db_name=mysettings.db_name
    )
config.set_main_option(name="sqlalchemy.url", value=connstring)

# add your model's MetaData object here
# for 'autogenerate' support
# from myapp import mymodel
# target_metadata = mymodel.Base.metadata
import db.initializer

target_metadata = db.initializer.Base.metadata

# other values from the config, defined by the needs of env.py,
# can be acquired:
# my_important_option = config.get_main_option("my_important_option")
# ... etc.


def run_migrations_offline() -> None:
    """Run migrations in 'offline' mode.

    This configures the context with just a URL
    and not an Engine, though an Engine is acceptable
    here as well.  By skipping the Engine creation
    we don't even need a DBAPI to be available.

    Calls to context.execute() here emit the given string to the
    script output.

    """
    url = config.get_main_option("sqlalchemy.url")
    context.configure(
        url=url,
        target_metadata=target_metadata,
        literal_binds=True,
        dialect_opts={"paramstyle": "named"},
    )

    with context.begin_transaction():
        context.run_migrations()


def run_migrations_online() -> None:
    """Run migrations in 'online' mode.

    In this scenario we need to create an Engine
    and associate a connection with the context.

    """
    connectable = engine_from_config(
        config.get_section(config.config_ini_section, {}),
        prefix="sqlalchemy.",
        poolclass=pool.NullPool,
    )

    with connectable.connect() as connection:
        context.configure(
            connection=connection, target_metadata=target_metadata
        )

        with context.begin_transaction():
            context.run_migrations()


if context.is_offline_mode():
    run_migrations_offline()
else:
    run_migrations_online()

Обновление конфигурации таблиц

Для проверки создаем первую пустую миграцию. После ее выполнения создастся таблица alembic_version

alembic revision -m "Empty Init"

В данный момент фактического соединения с БД не было. В папке versions сформируется файл вида <id>_empty_init.py 

После выполнения команды 

alembic upgrade head

в таблице alembic_version появится одна запись - идентификатор текущей версии базы данных.

Сейчас в папке db создаем папку tablesdefinition. В ней будем хранить файлы описаний таблиц и методы для работы с таблицами. Создадим файл userprofile.py 

'''Definition tables of userprofile'''
import logging
from sqlalchemy import Column, Integer, String
from sqlalchemy.orm import sessionmaker
import db.initializer
from db.initializer import engine

def create_userprofile_class(Curbase):
    class Userprofile(Curbase):
        '''Class Userprofile definition'''
        __tablename__ = 'userprofile'

        user_id = Column(Integer(), primary_key=True)
        username = Column(String(15), nullable=False, unique=True)
        password = Column(String(255), nullable=False)
        email = Column(String(255))
        name = Column(String(100))
        second_name = Column(String(100))
        photo = Column(String(255))
        balance = Column(Integer())
    return Userprofile

def create_one_userprofile(username, password, email='', name='',
                           second_name='', photo='', balance=0):
    ''' Create one userprofile '''
    try:
        with engine.connect():
            Session = sessionmaker(bind=engine)
            with Session() as sess:
                upelem = db.initializer.userprofile(username=username, password=password,
                                     email=email, name=name, second_name=second_name,
                                     photo=photo, balance=balance)
                sess.add(upelem)
                sess.commit()
    except Exception as e:
        logging.error(e)

И в файле initializer.py после инициализации переменной Base добавим раздел инициализации описания таблицы 

#============================ Creation classes definitions =========================
# === Import Userprofile class ===
from db.tablesdefinition.userprofile import create_userprofile_class
userprofile = create_userprofile_class(Base)
# ==================================================================================

Теперь после выполнения команды 

alembic revision --autogenerate -m "Added userprofile model"

будет автоматически сгененрирован файл миграции, и после

alembic upgrade head

создастся таблица userprofile.

P.s. В точке входа необходимо полностью импортировать initializer иначе будет ошибка, пример: 

'''Main learning module'''
import logging

from db import initializer
from db.tablesdefinition.userprofile import create_one_userprofile

logging.basicConfig(level=logging.INFO,
                    filename='error.log',
                    format="%(levelname)s %(message)s")

if __name__ == '__main__':
    create_one_userprofile(username = 'first6',
                           password = 'first',
                           balance = 1)

Alembic

Установка и настройка

Стандартная установка:

pip install alembic

Первая инициализация: alembic init folder_for_dbdata 

alembic init alembic

Будет создан файл alembic.ini и директория в соответствии с названием. 

В alembic.ini обновляем параметр sqlalchemy.url

В файл env.py добавляем код

import os
import sys
sys.path.append(os.getcwd())
from app.db import Base 
target_metadata = Base.metadata 

Миграции

Создание первой (пустой) миграции.

После создания пустой миграции, в БД создастся таблица alembic_version в которой хранится идентификатор текущей версии.

alembic revision -m "Empty Init"

Обновление базы данных после создания миграции

alembic upgrade head

Хэш текущей миграции

alembic current

История миграций.

Считываются все файлы миграций. Т.е. в случае загрузки предыдущего состояния, последующие миграции будут отображаться до тех пор, пока ненужные файлы миграций не будут удалены.

alembic history

Возврат к предыдущему состоянию

alembic downgrade migration_id

Для возврата в стартовое состояние, нужно выполнить 

alembic downgrade base

Если что-то пошло не так, для возврата в нулевое состояние нужно удалить из таблицы alembic_version текущий номер.

DELETE FROM public.alembic_version;

Пропуск состояния

alembic stamp migration_id

Экспорт в формате sql 

alembic upgrade migration_id_start:migration_id_stop --sql > migration.sql

Автогенерация миграции

alembic revision --autogenerate -m "Added Cookie model"

Поддерживаемые и неподдерживаемые действия при автоматической миграции

Чтобы alembic увидел класс данных, необходимо его непосредственно импортировать. Импорт всей директории не работает.

Ручное создание миграций

На примере изменения имени таблицы

• Изменить имя.
• Выполнить генерацию миграции

alembic revision -m "Renaming table"

• В созданном файле изменить upgrade/downgrade

def upgrade():
    op.rename_table('old_name', 'new_name')
def downgrade():
    op.rename_table('new_name', 'old_name')

• Выполнить миграцию

alembic upgrade head

Команды alembic

Модули

Описание модулей

Хранение конфигурации

Configparser стандартная библиотека для чтения и записи .ini файлов. Инструкция 1 

Jsonschema модуль для проверки соответствия json существующей схеме. Документация 

Libsettings модуль на основе jsonschema для чтения конфигурации из json файла и проверки конфигурации на соответствие схеме. Gitverse проекта

import logging
from libsettings import Jsettings

logging.basicConfig(level=logging.ERROR,
                    filename='error.log',
                    format="%(levelname)s %(message)s")
mysettings = Jsettings(settingsfname='mysettings.json',
                          schemafname='myschema.json')
mysettings.load_settings()

jsonschema

Используется для валидации json схемы. По умолчанию дополнительно указанные ключи (не существующие в схеме, но присутствующие в документе) не проверяются.

Установка

pip install jsonschema

Базовое использование

from jsonschema import validate

validate(instance=json_to_check, schema=schema)

Исключения

jsonschema.exceptions.ValidationError - если документ не соответствует структуре

jsonschema.exceptions.SchemaError - если сама схема некорректна

Пример схемы:

schema = {
    "type": "object",
    "properties": {
        "name": {"type": "string"},
        "age": {"type": "number"},
    },
    "required": ["name"],
}

Данная схема определяет json объект с 2 свойствами: name и age. Обязательное свойство name.

Ключевые слова

Для некоторых типов используются дополнительные ключевые слова.

Дополнительные ключевые слова для типов.

 Тип array

"scores": {
            "type": "array",
            "items": {"type": "number"},
        }

Тип object

"required": ["name"]

"properties": {
        "name": {"type": "string"}
    },

Локальные вложенные шаблоны.

Для определения используется переменная $defs, для использования - $ref. 

schema = {
    "type": "object",
    "properties": {
        "address": {"$ref": "#/$defs/address"},
    },
    "$defs": {
        "address": {
            "type": "object",
            "properties": {
                "street": {"type": "string"},
            }
        },
    },
}

Pydantic 2

Описание

Библиотека валидации (проверка на соответствие типов) и трансформации (автоматическое приведение к нужным типам и форматам) данных.

Модели наследуются от класса BaseModel. Модель описывает набор полей, представляющих структуру данных и условия  валидации.

Установка

pip install pydantic

Типизация:

• Простая: указание типа, например, name: str. 
• Объект Field(): дополнительные параметры, например, значения по умолчанию, ограничения и т.д.

Внутри класса можно комбинировать способы типизации. 

from pydantic import BaseModel, Field

class User(BaseModel):
    name: str
    email: str = Field(..., alias='email_address')

Валидация:

• Минимальная валидация: встроенные типы Python (например, str, int).
• Валидаторы: например EmailStr для проверки email-адресов. Требуется установка дополнительных зависимостей: pydantic[email] или pydantic[all].
• @field_validator — добавляет логику валидации поля. Вызывается при создании или изменении модели.

from pydantic import BaseModel, field_validator

class User(BaseModel):
    age: int

    @field_validator('age')
    def check_age(cls, value):
        if value < 18:
            raise ValueError('Возраст должен быть больше 18 лет')
        return value

• @computed_field — вычисляемое поле на основе данных в модели. Его можно использовать для автоматической генерации значений, а также для валидации.

from pydantic import BaseModel, computed_field

class User(BaseModel):
    name: str
    surname: str

    @computed_field
    def full_name(self) -> str:
        return f"{self.name} {self.surname}"

• @model_validator - валидация всей модели. 

Работает со всей моделью (а не с отдельными полями), может изменять данные перед валидацией (mode='before') или после (mode='after'), полезен для комплексных проверок, где одно поле зависит от другого, может возвращать новую версию модели (если нужно модифицировать данные).

@model_validator(mode='before')
def validate_before(cls, data: dict):
    if 'username' not in data:
        data['username'] = "guest_" + str(data.get('id', 0))
    return data

@model_validator(mode='after')
def validate_after(self):
    if self.age < 18 and self.is_premium:
        raise ValueError("Minors cannot have premium accounts!")
    return self

При проверке before передается класс, при after - объект. Можно делать два валидатора: before для подстановки вычисляемых значений и after для финальной проверки

Интеграция с SQLAlchemy:

Для настройки используется параметр ConfigDict с флагом from_attributes=True. 

from datetime import date
from pydantic import BaseModel, ConfigDict

class User(BaseModel):
    id: int
    name: str = 'John Doe'
    birthday_date: date

    config = ConfigDict(from_attributes=True)

Для создания модели Pydantic из объекта ORM используется метод from_orm. 

user = User.from_orm(orm_instance)

Ссылки:

Основа для текста

Rabbitmq

Базовая информация

Авторизация через ВК

Перейти на VK для разработчиков

Создать новое приложение, страницы настроек: 

В настройках приложения будут защищенный ключ и сервисный ключ.

Тестирование Playwright

Начало

Официальный сайт проекта

Установка

python -m pip install playwright

Проверка установки

playwright --version

Установка драйверов для браузеров 

playwright install #Все браузеры
playwright install name #Только name браузеры
playwright install chromium #Chrome

Установка pytest 

python -m pip install pytest

Установка плагина pytest-playwright 

python -m pip install pytest-playwright

Описание 

Два режима работы: синхронный и асинхронный. Для синхронного: 

from playwright.sync_api import sync_playwright

Для асинхронного режима: 

import asyncio
from playwright.async_api import async_playwright

async def main():
    async with async_playwright() as playwright:
        browser = await playwright.chromium.launch()
        page = await browser.new_page()
        await page.goto("https://playwright.dev")
        print(await page.title())
        await browser.close()

asyncio.run(main())

Чаще используется синхронный режим.

Запуск и закрытие браузера: 

from playwright.sync_api import sync_playwright

with sync_playwright() as playwright:
    browser = playwright.chromium.launch(headless=False, slow_mo=500)
    page = browser.new_page()
    page.goto("https://playwright.dev/python")

    docs_button = page.get_by_role('link', name="Docs")
    docs_button.click()

    browser.close()

headless=False обозначает визуальное открытие, slow_mo задержка

Использование интерактивной консоли

Иногда для удобства можно использовать консоль python для ручного тестирования покомандного ввода. 

python
Python 3.13.1 (tags/v3.13.1:0671451, Dec  3 2024, 19:06:28) [MSC v.1942 64 bit (AMD64)] on win32
Type "help", "copyright", "credits" or "license" for more information.
>>> from playwright.sync_api import sync_playwright
>>> playwright = sync_playwright().start()
>>> browser = playwright.chromium.launch(headless=False, slow_mo=100)
>>> page = browser.new_page()
>>> browser.close()
>>> playwright.stop()

Если создать docs_button, у нее будет метод highlight() для визуальной подсветки найденного элемента.

Локаторы

Локаторы: способ поиска элементов на странице. Поэтому они являются методами page

В VSC Ctrl+Click по методу выводит код метода.

<div>
  <label for="exampleInputEmail1" class="form-label mt-4">Email address</label>
  <input type="email" class="form-control" id="exampleInputEmail1" aria-describedby="emailHelp" placeholder="Enter email">
  <small id="emailHelp" class="form-text text-muted">We'll never share your email with anyone else.</small>
</div>

Доступ к iframe

'''Test for auth module'''
import pytest
from playwright.sync_api import Browser, Page, expect

AUTH_URL = "https://wood.bobrobotirk.ru/auth"

@pytest.fixture
def page_and_auth(browser: Browser):
    context = browser.new_context(
        storage_state="playwright/.auth/vk.json"
        )
    page = context.new_page()
    yield page
    context.close()

def test_first(page_and_auth: Page):
    page_and_auth.goto(AUTH_URL)
    vkframe = page_and_auth.frame(url=lambda url: "id.vk.com" in url)

    if vkframe:
        authbutton = vkframe.get_by_role("button", name="Продолжить как")
        expect(authbutton, "Кнопка Продолжить как... отсутствует").to_be_visible(timeout=20000)
        authbutton.click()
        back_button = page_and_auth.get_by_role("button", name="Авторизация успешна")
        expect(back_button, "Сервис не произвел авторизацию").to_be_visible()
        back_button.click()
    else:
        assert False, "VK фрейм не найден."

Actions

with page.expect_file_chooser() as fc_info:
    file_input.click() #до этого через локатор найден file_input
file_chooser = fc_info.value
file_chooser.set_files("first.txt")

События (Events)

События в page.goto

В переменной wait_until.

• load: загрузка всего контента
• domcontentloaded: загрузка dom
• commit: при получении ответа от сервера 
• networkidle: до завершения всех событий сети. Для динамического контента не меньше чем load.

Можно считать время загрузки. 

from playwright.sync_api import sync_playwright
from time import perf_counter

with sync_playwright() as playwright:
    browser = playwright.chromium.launch(headless=False, slow_mo=500)
    page = browser.new_page()
    print('Page loading...')
    start = perf_counter()
    page.goto("https://playwright.dev/python", wait_until='load')
    delta = perf_counter() - start
    print(f'Page loaded in {delta} s.')
    
    browser.close()

События динамического контента (React, ...)

Находим динамический элемент, кликаем по нему и при помощи wait_for() ждем.

from playwright.sync_api import sync_playwright
from time import perf_counter

with sync_playwright() as playwright:
    browser = playwright.chromium.launch(headless=False, slow_mo=500)
    page = browser.new_page()
    page.goto("https://www.scrapethissite.com/pages/ajax-javascript/", timeout=60_000)
    mylink = page.get_by_role("link", name="2014")
    mylink.click()
    print('Loading movie...')
    start = perf_counter()
    loadedcont = page.locator("td.film-title").first
    loadedcont.wait_for()
    delta = perf_counter() - start
    print(f'Movie loaded in {delta} s.')
    
    browser.close()

Ожидание события.

Указываем тип события и функцию, выполняемую при наступлении события. 

from playwright.sync_api import sync_playwright

def onload(page):
    print("Page loaded", page)

with sync_playwright() as playwright:
    browser = playwright.chromium.launch(headless=False, slow_mo=500)
    page = browser.new_page()
    page.on("load", onload)
    page.goto("https://bootswatch.com/default")
    browser.close()

Пример просмотра событий запросов 

from playwright.sync_api import sync_playwright

def onrequest(request):
    print("Request send: ", request)

with sync_playwright() as playwright:
    browser = playwright.chromium.launch(headless=False, slow_mo=500)
    page = browser.new_page()
    page.on("request", onrequest)
    page.goto("https://bootswatch.com/default")
    browser.close()

Вывод: 

Request send:  <Request url='https://bootswatch.com/default' method='GET'>
Request send:  <Request url='http://bootswatch.com/default/' method='GET'>
Request send:  <Request url='https://bootswatch.com/default/' method='GET'>
Request send:  <Request url='https://bootswatch.com/_vendor/bootstrap/dist/css/bootstrap.css' method='GET'>
Request send:  <Request url='https://bootswatch.com/_vendor/bootstrap-icons/font/bootstrap-icons.min.css' method='GET'>
Request send:  <Request url='https://bootswatch.com/_vendor/prismjs/themes/prism-okaidia.css' method='GET'>
Request send:  <Request url='https://bootswatch.com/_assets/css/custom.min.css' method='GET'>
Request send:  <Request url='https://www.googletagmanager.com/gtag/js?id=G-KGDJBEFF3W' method='GET'>
Request send:  <Request url='https://cdn.carbonads.com/carbon.js?serve=CKYIE23N&placement=bootswatchcom' method='GET'>
Request send:  <Request url='https://bootswatch.com/_vendor/bootstrap/dist/js/bootstrap.bundle.min.js' method='GET'>
Request send:  <Request url='https://bootswatch.com/_vendor/prismjs/prism.js' method='GET'>
Request send:  <Request url='https://bootswatch.com/_assets/js/custom.js' method='GET'>
Request send:  <Request url='https://bootswatch.com/_vendor/bootstrap-icons/font/fonts/bootstrap-icons.woff2?1fa40e8900654d2863d011707b9fb6f2' method='GET'>
Request send:  <Request url='https://www.google-analytics.com/g/collect?v=2&tid=G-KGDJBEFF3W&gtm=45je54g0v9135688085za200&_p=1744912703288&gcd=13l3l3l3l1l1&npa=0&dma=0&tag_exp=102509682~102803279~102813109~102887800~102926062~103027016~103051953~103055465~103077950~103106314~103106316~103130495~103130497&cid=128797986.1744912704&ul=ru-ru&sr=1280x720&uaa=x86&uab=64&uafvl=Not%253AA-Brand%3B24.0.0.0%7CChromium%3B134.0.6998.35&uamb=0&uam=&uap=Windows&uapv=10.0.0&uaw=0&are=1&frm=0&pscdl=noapi&_s=1&sid=1744912703&sct=1&seg=0&dl=https%3A%2F%2Fbootswatch.com%2Fdefault%2F&dt=Bootswatch%3A%20Default&en=page_view&_fv=1&_nsi=1&_ss=1&_ee=1&tfd=2018' method='POST'>
Request send:  <Request url='https://srv.carbonads.net/ads/CKYIE23N.json?segment=placement:bootswatchcom&v=true' method='GET'>

Есть событие при выборе файла.

Удаление прослушивания события: page.remove_listener("name_event", func)

События всплывающих окон (alert, confirm, prompt)

Событие page.on("dialog", func), устанавливаем ожидание до возможного появления окна.

Функция: 

def on_dialog(dialog)
    dialog.accept()
    dialog.dismiss()

Если диалог prompt, то для заполнения поля ввода в функции accept нужно добавить строковую переменную. 

def on_dialog(dialog)
    dialog.accept("Text for enter")

Событие скачивания файла

from playwright.sync_api import sync_playwright

with sync_playwright() as playwright:
    browser = playwright.chromium.launch(headless=False, slow_mo=500)
    page = browser.new_page()
    page.goto("https://bootswatch.com/default")
    btn = page.get_by_role("link", name="Download me")
    with page.expect_download() as download_info:
        btn.click()
    download = download_info.value
    download.save_as(fname)
    browser.close()

Второй вариант: добавить listener 

from playwright.sync_api import sync_playwright

def saving_func(download):
    fname = "first.jpg"
    download.save_as(fname)

with sync_playwright() as playwright:
    browser = playwright.chromium.launch(headless=False, slow_mo=500)
    page = browser.new_page()
    page.goto("https://bootswatch.com/default")
    page.once("download", saving_func)
    btn = page.get_by_role("link", name="Download me")
    with page.expect_download() as download_info:
        btn.click()
    browser.close()

Аутентификация

При 2FA аутентификации возникают проблемы при повторном исполнении скрипта. Для обхода этого используют контекст браузера. 

Шаг 1. Сохранение контекста. 

from playwright.sync_api import sync_playwright

with sync_playwright() as playwright:
    browser = playwright.firefox.launch(headless=False, slow_mo=500)
    context = browser.new_context()
    page = context.new_page()
    page.goto("https://vk.ru")
    page.pause() # Откроется доп. окно. Проходим авторизацию, в доп.окне play
    context.storage_state(path="playwright/.auth/vk.json")
    context.close()

Шаг 2. Использование контекста 

from playwright.sync_api import sync_playwright

with sync_playwright() as playwright:
    browser = playwright.firefox.launch(headless=False, slow_mo=500)
    context = browser.new_context(storage_state="playwright/.auth/vk.json")
    page = context.new_page()
    page.goto("https://vk.ru")
    page.pause() #в реальных скриптах это убирается)
    context.close()

Pytest & Playwright

Pytest

Имена файлов тестов должны иметь префикс test_ или постфикс _test. Имена тестов должны иметь префикс test_

В модуле utils функция root, отнимающая 1 от входного параметра. Пример теста: 

import utils

def test_first():
    num24 = utils.root(25)
    assert num24 == 24

Запуск теста: 

pytest second_test.py

Ключ -v отображает расширенную информацию.
Ключ -s разрешает вывод данных из тестируемых функций. 

Запуск без указания имени файла исполняет все тесты.

Желательно определение типов в функциях.

Виды проверок

Для вывода доп. информации: 

assert mynum == 32, "Число mynum должно быть равно 32"

Фикстуры:

Для упаковки повторяющихся действий. 

import json
import report
import pytest

@pytest.fixtures
def report_json():
    report.generate_report()
    with open("report.json") as file:
        return json.load(file)

def test_report_json(report_json):
    assert type(report_json) == dict

Фикстура исполняется каждый раз при вызове. Для однократного исполнения фикстуры нужно добавить scope="session" 

import json
import report
import pytest

@pytest.fixtures(scope="session")
def report_json():
    report.generate_report()
    with open("report.json") as file:
        return json.load(file)

def test_report_json(report_json):
    assert type(report_json) == dict

Виды scope

Pytest-playwright

Данный плагин упрощает работу с pytest. Встроенные фикстуры. Задачу создания и удаления playwright, browser берет на себя.

from playwright.sync_api import Page

def test_first(page: Page):
    page.goto("https://playwright.dev/python")
    link = page.get_by_role("link", name="GET STARTED")
    link.click()
    assert page.url == "https://playwright.dev/python/docs/intro"

Настройка pytest

Через консоль:

• --headed Отображать окно
• --slowmo=500  Замедление работы
• --browser=firefox  Настройка браузера

Через конфигурационный файл pytest.ini

[pytest]
addopts = --headed --slowmo=500 --browser=firefox

Фикстуры

Добавить фикстуру 

import pytest
from playwright.sync_api import Page

@pytest.fixture
def load_testpage(page: Page):
    page.goto("https://playwright.dev/python")
    return page

def test_first(load_testpage: Page):
    link = load_testpage.get_by_role("link", name="GET STARTED")
    link.click()
    assert load_testpage.url == "https://playwright.dev/python/docs/intro"

Автоматическое исполнение фикстуры перед каждой функцией:

import pytest
from playwright.sync_api import Page

@pytest.fixture(autouse=True)
def load_testpage(page: Page):
    page.goto("https://playwright.dev/python")
    return page

def test_first(page: Page):
    link = page.get_by_role("link", name="GET STARTED")
    link.click()
    assert page.url == "https://playwright.dev/python/docs/intro"

Пред и пост исполнение кода:

import pytest
from playwright.sync_api import Page

@pytest.fixture(autouse=True)
def load_testpage(page: Page):
    page.goto("https://playwright.dev/python")
    yield page
    page.goto("https://yandex.ru")

def test_first(page: Page):
    link = page.get_by_role("link", name="GET STARTED")
    link.click()
    assert page.url == "https://playwright.dev/python/docs/intro"

Пример проверки авторизации ВК 

'''Test for auth module'''
import pytest
from playwright.sync_api import Browser, Page, expect

AUTH_URL = "https://wood.bobrobotirk.ru/auth"

@pytest.fixture
def page_and_auth(browser: Browser):
    context = browser.new_context(
        storage_state="playwright/.auth/vk.json"
        )
    #page = context.new_page()
    yield context
    context.close()

def get_cookie(all_cookies, cname):
    val = next((cookie['value'] for cookie in all_cookies if cookie.get('name') == cname), None)
    return val

def test_first(page_and_auth: Browser):
    page = page_and_auth.new_page()
    page.goto(AUTH_URL)
    vkframe = page.frame(url=lambda url: "id.vk.com" in url)

    if vkframe:
        authbutton = vkframe.get_by_role("button", name="Продолжить как")
        expect(authbutton, "Кнопка Продолжить как... отсутствует").to_be_visible(timeout=20000)
        authbutton.click()
        back_button = page.get_by_role("button", name="Авторизация успешна")
        expect(back_button, "Сервис не произвел авторизацию").to_be_visible()
        back_button.click()
        all_cookies = page_and_auth.cookies()
        session_id_value = get_cookie(all_cookies, 'session_id')
        assert session_id_value, "Cookie session_id не установлен."
    else:
        assert False, "VK фрейм не найден."

Дополнительные возможности

Скриншоты

Скрин страницы

page.screenshot(path="", full_page=True)

Скрин элемента тоже работает.

link.screenshot(path="")

Запись видео

from playwright.sync_api import Browser

def test_first(browser: Browser):
    context = browser.new_context(
        storage_state="playwright/.auth/vk.json",
        record_video_dir="video/"
        )
    page = context.new_page()
    page.goto("https://playwright.dev/python")
    link = page.get_by_role("link", name="GET STARTED")
    link.click()
    assert page.url == "https://playwright.dev/python/docs/intro"
    page.goto("https://vk.ru")

Вариант с текстурой: 

import pytest
from playwright.sync_api import Browser, Page

@pytest.fixture
def recordable(browser: Browser):
    context = browser.new_context(
        storage_state="playwright/.auth/vk.json",
        record_video_dir="video/"
        )
    page = context.new_page()
    yield page
    context.close()

def test_first(recordable: Page):
    recordable.goto("https://playwright.dev/python")
    link = recordable.get_by_role("link", name="GET STARTED")
    link.click()
    assert recordable.url == "https://playwright.dev/python/docs/intro"
    recordable.goto("https://vk.ru")

Трассировка данных

Создание трассировки: 

import pytest
from playwright.sync_api import Page, BrowserContext

@pytest.fixture(autouse=True)
def trace_test(context: BrowserContext):
    context.tracing.start(
        name="playwrite",
        screenshots=True,
        snapshots=True,
        sources=True
    )
    yield 
    context.tracing.stop(path="trace.zip")

def test_first(page: Page):
    page.goto("https://playwright.dev/python")
    link = page.get_by_role("link", name="GET STARTED")
    link.click()
    assert page.url == "https://playwright.dev/python/docs/intro"

Просмотр трассировки 

playwright show-trace trace.zip

Запись действий

Старт записи 

playwright codegen playwright.dev

В окне кода есть поле Target, позволяет переключать тип библиотек (playwright sync/async, pytest). Потом сохраняем полученную последовательность и устанавливаем нужные условия для проверки.

Ожидание

from playwright.sync_api import Page, expect

DOCS_URL = "https://playwright.dev/python/docs/intro"

def test_first(page: Page):
    page.goto("https://playwright.dev/python")
    link = page.get_by_role("link", name="GET STARTED")
    link.click()
    #assert page.url == DOCS_URL
    expect(page).to_have_url(DOCS_URL)

expect(mylink).to_have_class(
    re.compile(r"navbar__link")
    )

not_ - префикс отрицания