Installation

pip install --extra-index-url https://pypi.abm-jsc.ru sqlalchemy-tools-abm

Quick Start

Create simple entity

from sqlalchemy import Integer, String
from sqlalchemy_tools.entity_helpers.decorator import sqlalchemy_dataclass
from sqlalchemy_tools.entity_helpers.fields import sql_field
from sqlalchemy_tools.entity_helpers.setter import set_ids
from sqlalchemy_tools.entity_helpers.sqlalchemy_base import sqlalchemy_mapper_registry, register_initial_values


@sqlalchemy_dataclass(sqlalchemy_mapper_registry)
class Language:
    id: int = sql_field(Integer, primary_key=True)
    code: str = sql_field(String, default=None)


languages = [Language(code="ru"), Language(code="en")]
set_ids(*languages)
language_id_to_code = {language.id: language.code for language in languages}

register_initial_values(
    sqlalchemy_mapper_registry,
    *languages
)

Create entity with relationship

import time

from typing import List
from sqlalchemy import Integer, String, ForeignKey, BigInteger
from sqlalchemy_tools.entity_helpers.decorator import sqlalchemy_dataclass
from sqlalchemy_tools.entity_helpers.fields import sql_field, sql_relation_field
from sqlalchemy_tools.entity_helpers.sqlalchemy_base import sqlalchemy_mapper_registry

@sqlalchemy_dataclass(sqlalchemy_mapper_registry)
class CarouselImage:
    id: int = sql_field(Integer, primary_key=True)

    file_id: int = sql_field(Integer, nullable=False)
    file_name: str = sql_field(String, nullable=False)

    widget_id: int = sql_field(Integer, ForeignKey("widget.id", ondelete='CASCADE'))

@sqlalchemy_dataclass(sqlalchemy_mapper_registry)
class Widget:
    id: int = sql_field(Integer, primary_key=True)
    title: str = sql_field(String, default="")
    created_at: int = sql_field(BigInteger, default=int(time.time() * 1000))

    carousel_images: List[CarouselImage] = sql_relation_field(CarouselImage, use_list=True)

Add

from sqlalchemy_tools.database_connector.database_connector import DatabaseConnector
from sqlalchemy_tools.sqlalchemy_crud.sqlalchemy_crud import SQLAlchemyCRUD


async def add():
    config = DatabaseConnector.Config(address="postgresql://user:password@address/my_db")
    database_connector = DatabaseConnector(config=config)

    async with database_connector.get_session() as database_session:
        widget_ids = await SQLAlchemyCRUD.add(
            entity=Widget, 
            column_name_to_value_list=[
                {"created_at": 1, "title": "title1", "carousel_images": [{"file_id": 1, "file_name": "file_name1"}]},
                {"title": "title2", "carousel_images": [{"file_id": 2, "file_name": "file_name2"}, 
                                                        {"file_id": 3, "file_name": "file_name3"}]}
            ], 
            db_session=database_session
        )
        await database_session.commit()

Get

Get method  hasn't been implemented yet.

Update

from sqlalchemy_tools.database_connector.database_connector import DatabaseConnector
from sqlalchemy_tools.sqlalchemy_crud.sqlalchemy_crud import SQLAlchemyCRUD


async def update():
    config = DatabaseConnector.Config(address="postgresql://user:password@address/my_db")
    database_connector = DatabaseConnector(config=config)

    async with database_connector.get_session() as database_session:
        updated_status = await SQLAlchemyCRUD.update(
            entity=Widget,
            column_name_to_value_list=[
                {"id": 1, "title": "new_title1", "carousel_images": [{"id": 1, "file_id": 2, "file_name": "new_name"}]},
                {"id": 2, "title": "new_title2"}
            ],
            db_session=database_session
        )
        await database_session.commit()

Delete

from sqlalchemy_tools.database_connector.database_connector import DatabaseConnector
from sqlalchemy_tools.sqlalchemy_crud.sqlalchemy_crud import SQLAlchemyCRUD


async def delete():
    config = DatabaseConnector.Config(address="postgresql://user:password@address/my_db")
    database_connector = DatabaseConnector(config=config)

    async with database_connector.get_session() as database_session:
        deleted_status = await SQLAlchemyCRUD.delete(
            entity=Widget,
            filters=[Widget.id == 1],
            db_session=database_session
        )
        await database_session.commit()