miguelgrinberg/alchemical

SQLAlchemy 2.0+ wrapper that simplifies its use in Python applications. Can be used on its own or alongside Flask, FastAPI or other web frameworks.

alchemical's Issues

A call to initialize() should fully reset the internal state

When initialize() is called a second time on an Alchemical object, the internal state should be reset, so that the next time it is accessed the engine(s) are recreated using the new settings.

Documentation is not clear

session.scalars(query): Returns an iterable with a single result per row.
session.scalar(query): Returns the first result from the first row.
session.execute(query): Returns an iterable with a tuple of results per row.

Did you mean to say something like
'session.scalars(query).all(): Returns an iterable with a single result per row.'?

How to set metadata (migrating from Flask-SQLAlchemy)?

Sorry for possibly stupid question.
I'm trying to migrate from Flask-SQLAlchemy (in hope of solving some errors, and also forcing me to use new querying syntax).

I was creating db this way:

db = SQLAlchemy(
    session_options={"autoflush": False},
    metadata=MetaData(naming_convention=convention),
)

and I'm not sure how to change this to Alchemical.
Is it correct to change session_options to engine_options?
And how do I give my naming conventions somewhere for sqlalchemy to use?

Thank you for help!

Check if the table exists

In Alchemical library is there a way to know if the table in database already exists?
if yes how?

Why do i need this functionality: i was trying to initiate the tables with "flask db upgrade" that didnt not create the tables. so had to use "db.create_all()" in code.
i dont want db.create_all() to be created again after running it, so better to check if table does not exists then create it.

i checked in sqlalchemy this functionality exists (link)

docs is not reachable

What is this?

Alternative for Flask-SQLAlchemy?

run an arbitrary SQL statement on the database not accurate

Hola Miguel,

Documentation indicated use db.text() however attribute does not exist.

Here is python modules in my enviroment.

How do I track session changes with events?

Multiple Database documentation.

I like the new way (in example files) of doing multiple databases more than the documentation on https://alchemical.readthedocs.io/en/latest/usage.html
Its a lot closer to the SQLAlchemy standard.
config.py

    ALCHEMICAL_DATABASE_URL = os.environ.get('DATABASE_URL') or \
        'sqlite:///' + os.path.join(basedir, 'db.sqlite')
    ALCHEMICAL_BINDS = {'firebird': 'firebird+fdb://sysdba:masterkey@localhost:3050/' + basedir + '\\DATA.FDB'}

Thankyou, no real issue, just posting here for anybody else who is currently working with multiple databases.

alchemical.aio.Alchemical with Flask and Flask-Migrate

Hey Miguel 👋 I hope all is well! I'm wondering if I could use Flask, and the as alchemical.aio.Alchemical object with Flask-Migrate. This seems like a combination that is probably not possible but I hope I'm wrong. For a somewhat artificial reason, I'm happy with uwsgi as my application server, and I don't want to switch over to aioflask/asgi/etc just yet. I've discovered some features in uwsgi that are keeping me there (for now). I'm interested in being able to get migrations, and utilize the new async support in Flask. Alchemical seems to get me really close! Curious to hear your thoughts on this, any advice appreciated.

Automatically generate migrations

I am currently trying to get the auto-generate to work. I have orientated myself on your examples in the retrofun repo.

However, the migration files are always empty and look like this:

from alembic import op
import sqlalchemy as sa


# revision identifiers, used by Alembic.
revision = '27b94fadbd77'
down_revision = None
branch_labels = None
depends_on = None


def upgrade(engine_name: str) -> None:
    globals()["upgrade_%s" % engine_name]()


def downgrade(engine_name: str) -> None:
    globals()["downgrade_%s" % engine_name]()


def upgrade_() -> None:
    pass


def downgrade_() -> None:
    pass

I proceeded as described in the documentation and first initialized with python -m alchemical.alembic.cli init migrations alembic and then tried to create a migration with alembic revision --autogenerate -m "Initial migration".

I have also adjusted the alchemical_db variable accordingly.

Am I missing something?

Listening for database events in Alchemical

Before going to my question, first I would like to thank you for your work, it is really amazing! I've watched/ read most of your tutorials and learned so much! It is incredible how you make complex topics feel so easy.

Now, let's start with the question. I am trying to implement a search functionality using Flask, react, apifairy, and Alchemical. In your mega flask tutorial, you add listeners (after_commit and before_commit) to the database in order to keep the elasticsearch indices and the db instance in sync. However, in the mega flask tutorial, you use SQLAlchemy, and I am using Alchemical. Is it possible to listen for events with alchemical?

Any help is appreciated :)

Inheritance mapping: [single-table, joined-table, and concrete-table] does not seem to work.

Greetings,

Attempting to port an application to Alchemical, I seem to have stumbled upon a peculiar case. The application in question uses Single-Table Inheritance for some models and Alchemical does not seem to be able to load them and breaks with the following error:

sqlalchemy.exc.InvalidRequestError: Row with identity key (<class '__main__.Base'>, (1,), None) can't be loaded into an object; the polymorphic discriminator column 'children.type' refers to mapped class B->children, which is not a sub-mapper of the requested mapped class A->children

At first I thought the bug resided in the application, but after some investigations, it really seems to be boiling down to Alchemical. Consider the following example:

Using:

• Alchemical: 0.5.1
• SQLAlchemy: 1.4.24
• Python: 3.10.5

Minimal example with vanilla SQLAlchemy:

import os
from sqlalchemy import Column, Integer, String, ForeignKey, create_engine
from sqlalchemy.ext.declarative import declared_attr
from sqlalchemy.orm import relationship, Session, declarative_base

Model = declarative_base()

class Parent(Model):
    __tablename__ = "parents"

    id = Column(Integer, primary_key=True)
    title = Column(String, nullable=True)

    all_a = relationship("A", backref="parent", foreign_keys="A.parent_id", cascade="all, delete-orphan")
    all_b = relationship("B", backref="parent", foreign_keys="B.parent_id", cascade="all, delete-orphan")


class Base(Model):
    __tablename__ = "children"

    id = Column(Integer, primary_key=True)
    title = Column(String, nullable=True)
    type = Column(String, nullable=False)

    @declared_attr
    def parent_id(self):
        return Column(Integer, ForeignKey("parents.id"), nullable=False)

    __mapper_args__ = {"polymorphic_on": type, "polymorphic_identity": "BASE"}


class A(Base):
    __mapper_args__ = {"polymorphic_identity": "A"}


class B(Base):
    __mapper_args__ = {"polymorphic_identity": "B"}


engine = create_engine(f"sqlite:///{os.path.join(os.getcwd(), 'database.db')}", echo=False, future=True)
Model.metadata.create_all(engine)
session = Session(engine)

p = Parent(title="Parent 1")
p.all_a.append(A(title="Child A1"))
p.all_b.append(B(title="Child B1"))

session.add(p)
session.commit()
session.close()

session = Session(engine)
p = session.get(Parent, 1)
print("-> ", p.title)
print("a: ", p.all_a)
print("b: ", p.all_b)

The result should look somewhat similar to this:

->  Parent 1
a:  [<__main__.A object at 0x7***>]
b:  [<__main__.B object at 0x7***>]

The same code adapted to using Alchemical:

import os
from sqlalchemy import Column, Integer, String, ForeignKey
from sqlalchemy.ext.declarative import declared_attr
from sqlalchemy.orm import relationship

db = Alchemical(url=f"sqlite:///{os.path.join(os.getcwd(), 'database.db')}")

class Parent(db.Model):
    __tablename__ = "parents"

    id = Column(Integer, primary_key=True)
    title = Column(String, nullable=True)

    all_a = relationship("A", backref="parent", foreign_keys="A.parent_id", cascade="all, delete-orphan")
    all_b = relationship("B", backref="parent", foreign_keys="B.parent_id", cascade="all, delete-orphan")


class Base(db.Model):
    __tablename__ = "children"

    id = Column(Integer, primary_key=True)
    title = Column(String, nullable=True)
    type = Column(String, nullable=False)

    @declared_attr
    def parent_id(self):
        return Column(Integer, ForeignKey("parents.id"), nullable=False)

    __mapper_args__ = {"polymorphic_on": type, "polymorphic_identity": "BASE"}


class A(Base):
    __mapper_args__ = {"polymorphic_identity": "A"}


class B(Base):
    __mapper_args__ = {"polymorphic_identity": "B"}

db.create_all()

with db.Session() as session:
    p = Parent(title="Parent 1")
    p.all_a.append(A(title="Child A1"))
    p.all_b.append(B(title="Child B1"))

    session.add(p)
    session.commit()
    session.close()

with db.Session() as session:
    p = session.get(Parent, 1)
    print("-> ", p.title)
    print("a: ", p.all_a)
    print("b: ", p.all_b)

Alchemical code produces the following result:

->  Parent 1
Traceback (most recent call last):

[... amended for brevity ...]

sqlalchemy.exc.InvalidRequestError: Row with identity key (<class '__main__.Base'>, (2,), None) can't be loaded into an object; the polymorphic discriminator column 'children.type' refers to mapped class B->children, which is not a sub-mapper of the requested mapped class A->children

Am I missing something? I went through the documentation several times but did not find anything relating to table inheritance or polymorphic relationships, so I presume it works identically to the vanilla SQLAlchemy.

The example above is an adapted version of the single-table inheritance as demonstrated in the SQLAlchemy docs.

[Help]Configure Alchemical for testing

Hi Miguel,
sorry to write this kind of issue here, but since there are no discussions set, here I go. My issue is to find a correct way to set Alchemical object in both code and tests. Do you know resources/examples to configure properly Alchemical?

For my use case I set up configuration using Pydantic v1 model via environment variables. This is where I set the connection string.

config.py module

from pydantic import BaseSettings, PostgresDsn


class Configuration(BaseSettings):
    db_connection_string: PostgresDsn | str = 'sqlite+pysqlite:///:memory:'

    class Config:
        env_prefix = 'ar_'

And I have a models.py module where I declare the Alchemical object and database models

from .config import Configuration

db = Configuration().db_connection_string


class Job(db.Model):
    __tablename__ = 'jobs'

    id: Mapped[uuid.UUID] = mapped_column(Uuid, primary_key=True, default=uuid.uuid4)
    name: Mapped[str] = mapped_column(String(50))
   ...

And now in my conftest.py module (I'm using Pytest for tests), I tried to set an environment variable to configure the connection string differently

@pytest.fixture(autouse=True)
def db_session(monkeypatch, mocker) -> Session:
    """Creates and deletes a test database."""
    monkeypatch.setenv('ar_db_connection_string', 'sqlite+pysqlite:///:memory:')
 
    # ugly but necessary
    from project.models import db

    db.create_all()
    with db.Session() as session:
        yield session
    db.drop_all()

This does not work because the configuration object is set before entering the fixture (I saw it with some breakpoints). I have no clue how to organize it differently. If you have the slightest suggestion, I'm interested.