Where communities thrive


  • Join over 1.5M+ people
  • Join over 100K+ communities
  • Free without limits
  • Create your own community
People
Activity
    Guilherme Carvalho
    @guicarvalho
    Top! self_group it worked :D... Tks you so much @CaselIT
    Federico Caselli
    @CaselIT
    btw, I think it's by design like this, since the parens are not actually needed https://www.postgresql.org/docs/current/sql-syntax-lexical.html#SQL-PRECEDENCE
    so the query without parens is ok in sql
    it sure is strange :)
    Alex Frase
    @atfrase:matrix.org
    [m]
    Hello! I'm new to SA and working on using a metaclass to generate the Table object for each table class, and then use "imperative mapping" to bind it to the class. In the docs I see that I have to call map_imperatively on a specific registry instance from sqlalchemy.orm.registry(), and I see that calling that generator twice yields different registry objects (so not a singleton); do I need to make sure all my table classes are mapped to the same registry object? What is that registry used for exactly, other than (I guess) detection of duplicate names?
    Alex Frase
    @atfrase:matrix.org
    [m]
    One other question: is there a way to get my metaclass to attach an event listener to table subclasses that will fire when any instrumented attribute is set, and can determine which attribute it was? So far I can have the metaclass __new__() run cls=super().__new__(...) ; for col in column_list: listen(getattr(cls,col), 'set', onset_handler, retval=True) and this does indeed trigger onset_handler() when setting any attribute of an instance of a subclass, but the event callback signature for 'set' only includes the new value and not the name of the field being set. It looks like maybe I can parse it out of initiator.impl or initiator.parent_token but I don't see much documentation on the AttributeEvent object that comes through as initiator, so I'm dubious if that strategy will hold up.
    (for completeness, I'll note that the full callback signature is receive_set(target, value, oldvalue, initiator) but target points to the class itself, not the specific instrumented attribute of the class that's currently being set)
    Alex Grönholm
    @agronholm
    @atfrase:matrix.org the registry is used for relationship mapping
    Cyril Scetbon
    @cscetbon
    Does someone know what needs to be done to migrate from flask-sqlalchemy to sqlalchemy ? We don't use Flask anymore and I'd like to be sure I do it correctly
    Federico Caselli
    @CaselIT
    I think mostly import from sqlalchemy and sqlalchemy.orm instead of flask db
    16 replies
    Alex Frase
    @atfrase:matrix.org
    [m]
    Thanks @agronholm , I'd guess that means I should take care to use the same registry for tables in the same database
    Alex Frase
    @atfrase:matrix.org
    [m]
    Next: Is there any better way to retrieve the Column() object that's mapped to a class attribute from inside @listens_for(attribute, 'set') ? The best I can come up with so far is to loop over getattr(initiator.impl.class_, initiator.impl.key).property.columns but that feels a little roundabout (AttributeImpl -> InstrumentedAttribute -> ColumnProperty -> Column)
    gnubyte
    @gnubyte
    Hi all.
    Im trying to dynamically create a field name in SQL-ALchemy where the field name is using part of a field from the model its being used in. Is there a good way to do this that anyone knows of? Im having a really hard time doing this with the SQLAlchemy events
    class myMixin:
       #dynamic fieldname here
       # {nameOfBook}_inherited = Column(String)
    
    class Books(myMixin):
      id =...
      nameOfBook = ...
    Alex Frase
    @atfrase:matrix.org
    [m]
    @gnubyte I'm not sure I follow exactly where your derived field name should come from, but I'd suggest looking into the "imperative mapping style" instead of declarative. In that scheme you'd create your own Table object including any number of Column objects, and you can name the Columns in their constructor using whatever logic you need to define the field name. Then you call another function to "map" the Table object to your class, whose definition could be empty (all the column properties will be created during mapping).
    gnubyte
    @gnubyte
    @atfrase:matrix.org could that be used in a mixin?
    Alex Frase
    @atfrase:matrix.org
    [m]
    Maybe? I did something similar (for a different purpose) the other day using a metaclass, where the meta __new__() took care of generating the Table object, then called super().__new__() to get the subclass created, and then called map_imperatively() to join them
    I'm not sure if a mixin could do it or not, I haven't used them much, but the reason I did it with a metaclass was that I needed the mapping to happen when the subclass was defined (which is like an instantiation of the metaclass, hence __new__()). I don't know if mixins have a place to run code at the time of subclass definition
    gnubyte
    @gnubyte
    Thats exactly what Ive been doing for about the last hour lol, minus the metaclass. Let me try that metaclass. Your explanation makes a lot of sense and in regards to mixins Im a bit concerned about the order of events.
    Xucong Zhan
    @HymanZHAN

    Hi, I am trying out the lambda_stmt function within the async example provided in the doc, but I am noticing the method add_criteria is not working for options(selectinload(A.bs)), but the += operator is working fine. Here is the code:

    """Illustrates use of the sqlalchemy.ext.asyncio.AsyncSession object
    for asynchronous ORM use.
    
    """
    
    import asyncio
    from operator import or_
    from typing import Callable, List
    
    from sqlalchemy import (
        Column,
        DateTime,
        ForeignKey,
        Integer,
        String,
        func,
        select,
        lambda_stmt,
    )
    from sqlalchemy.ext.asyncio import AsyncSession, create_async_engine
    from sqlalchemy.ext.declarative import declarative_base
    from sqlalchemy.orm import relationship, selectinload, sessionmaker
    from sqlalchemy.sql.selectable import Select
    
    Base = declarative_base()
    
    
    class A(Base):
        __tablename__ = "a"
    
        id = Column(Integer, primary_key=True)
        data = Column(String)
        create_date = Column(DateTime, server_default=func.now())
        bs = relationship("B")
    
        # required in order to access columns with server defaults
        # or SQL expression defaults, subsequent to a flush, without
        # triggering an expired load
        __mapper_args__ = {"eager_defaults": True}
    
        def __repr__(self) -> str:
            return f"<A Id:{self.id} Data:{self.data}>"
    
    
    class B(Base):
        __tablename__ = "b"
        id = Column(Integer, primary_key=True)
        a_id = Column(ForeignKey("a.id"))
        data = Column(String)
    
        def __repr__(self) -> str:
            return f"<B Id:{self.id} Data:{self.data}>"
    
    
    def order_by_id(klass: Base):
        return klass.id.desc()
    
    
    async def async_main():
        """Main program function."""
    
        engine = create_async_engine(
            "postgresql+asyncpg://test:testpassword@localhost/sqlalchemy_test",
            echo=True,
        )
    
        async with engine.begin() as conn:
            await conn.run_sync(Base.metadata.drop_all)
            await conn.run_sync(Base.metadata.create_all)
    
        # expire_on_commit=False will prevent attributes from being expired
        # after commit.
        async_session = sessionmaker(engine, expire_on_commit=False, class_=AsyncSession)
    
        async with async_session() as session:
            async with session.begin():
                session.add_all(
                    [
                        A(bs=[B(), B()], data="a1"),
                        A(bs=[B(data="a2b1")], data="a2"),
                        A(bs=[B(data="a3b1"), B(data="a3b2")], data="a3"),
                    ]
                )
    
            stmt = lambda_stmt(lambda: select(A))
            filter_by_id: Callable[[Select], Select] = lambda s: s.where(A.id > 1)
            order_by_id_desc: Callable[[Select], Select] = lambda s: s.order_by(A.id.desc())
            load_bs: Callable[[Select], Select] = lambda s: s.options(selectinload(A.bs))
    
            stmt.add_criteria(filter_by_id)
            stmt.add_criteria(order_by_id_desc)
            # stmt += load_bs # this works
            stmt.add_criteria(load_bs) # this doesn't work
            print("--------")
            print(stmt)
            print("--------")
    
            result = await session.execute(stmt)
            all_as: List[A] = result.scalars().all()
            print("--------")
            print(all_as)
            for a in all_as:
                for b1 in a.bs:
                    print(b1)
            print("--------")
    
    
    asyncio.run(async_main())

    Any help is appreciated. Thanks in advance!

    Federico Caselli
    @CaselIT
    @HymanZHAN I suggest you open a discussion
    mike bayer
    @zzzeek
    @HymanZHAN "stmt = stmt.add_criteria()" it does not mutate in place
    Federico Caselli
    @CaselIT
    do the other two statement not work also?
    mike bayer
    @zzzeek
    they should not
    Xucong Zhan
    @HymanZHAN
    Ah you are right. The other two criteria did not take effect either. Thanks a lot!
    Yannick
    @YannickLeRoux

    Got a newbie question, what would be the proper query for this?
    I got a table full of "settings", each row has a key column and a version column.
    So multiple rows will have the same key (string) but not the same version (int).
    How can I query all the rows with the latest version of each key.
    For example, I have a table like:

    • key: "aa", version:1
    • key: "aa", version:2
    • key: "aa", version:3
    • key: "bb", version:1

    And I want to have returned:

    • key: "aa", version:3
    • key: "bb", version:1

    I have seen func.max() that seems like a good idea but I am not sure how to put the query together.
    Thanks in advance.

    9 replies
    Mehdi ABAAKOUK
    @sileht
    Hi there, I'm replacing the sqlalchemy-stubs by the official one and I have a couple of question:
    I have a query with filter(gh_account.GitHubAccount.id % worker_count == worker_id)
    gh_account.GitHubAccount.id is Mapped[int], worker_count: int and worker_id: int
    mypy complain with: Non-overlapping equality check (left operand type: "ClauseElement", right operand type: "int")
    Any idea ?
    Maico Timmerman
    @MaicoTimmerman
    Currently, all python operators simplify expressions to ClauseElement, there is no other solution than adding a # type: ignore
    Mehdi ABAAKOUK
    @sileht
    Cool, I did nothing wrong :p
    Maico Timmerman
    @MaicoTimmerman
    The stubs are still WIP, we hope to be able to improve them to work with more complex ORM statements
    Mehdi ABAAKOUK
    @sileht
    The other one I have:
    class MyEnum(enum.Enum):
        ...
    
    kind: sqlalchemy.orm.Mapped[MyEnum] = sqlalchemy.Column(sqlalchemy.Enum(MyEnum), nullable=False)
    Mypy complains and said I should use Mapped[str]
    Does this ring a bell ?
    Mehdi ABAAKOUK
    @sileht
    Left hand assignment 'type: "MyEnum"' not compatible with ORM mapped expression of type "Mapped[str]" [misc]
    Maico Timmerman
    @MaicoTimmerman
    I think we should be able to that with Mapped[MyEnum] when you have the plugin enabled
    Mehdi ABAAKOUK
    @sileht
    This is what I do, and I have the plugin enabled
    Maico Timmerman
    @MaicoTimmerman
    What version of SQLAlchemy 1.4 are you running? There have been some fixes with enums, so make sure you are running the latest
    Mehdi ABAAKOUK
    @sileht
    1.4.13
    Maico Timmerman
    @MaicoTimmerman
    You then might be running into a bug, can you open an issue with a MCVE on github?
    Mehdi ABAAKOUK
    @sileht
    Yes

    And the last one I have:

    created_at: sqlalchemy.orm.Mapped[datetime.datetime] = sqlalchemy.Column(
         sqlalchemy.DateTime, server_default=sqlalchemy.func.now(), nullable=False
    )

    raise:

    error: No overload variant of "Column" matches argument types "Type[DateTime]", "Function[NullType]", "bool"  [call-overload]
    note: Possible overload variants:
    note:     def [_TE <: TypeEngine[Any]] __init__(self, str, *args: SchemaEventTarget, autoincrement: Union[bool, Literal['auto'], Literal['ignore_fk']] = ..., default: Optional[Any] =
    ..., doc: Optional[str] = ..., key: Optional[str] = ..., index: Optional[bool] = ..., info: Mapping[Any, Any] = ..., nullable: bool = ..., onupdate: Optional[Any] = ..., primary_key: bool = ..., server_default: Union[FetchedValue, str, TextClause, None] = ..., server_onupdate: Optional[FetchedValue] = ..., quote: Optional[bool] = ..., unique: Optional[bool] = ..., system: bool = ..., comment: Optional[str] = ..., **kwargs: Any) -> Column[NullType]
    note:     def [_TE <: TypeEngine[Any]] __init__(self, *args: SchemaEventTarget, autoincrement: Union[bool, Literal['auto'], Literal['ignore_fk']] = ..., default: Optional[Any] = ...,
    doc: Optional[str] = ..., key: Optional[str] = ..., index: Optional[bool] = ..., info: Mapping[Any, Any] = ..., nullable: bool = ..., onupdate: Optional[Any] = ..., primary_key: bool = ..., server_default: Union[FetchedValue, str, TextClause, None] = ..., server_onupdate: Optional[FetchedValue] = ..., quote: Optional[bool] = ..., unique: Optional[bool] = ..., system: bool = ..., comment: Optional[str] = ..., **kwargs: Any) -> Column[NullType]
    note:     def [_TE <: TypeEngine[Any]] __init__(self, Union[_TE, Type[_TE]], *args: SchemaEventTarget, autoincrement: Union[bool, Literal['auto'], Literal['ignore_fk']] = ..., default: Optional[Any] = ..., doc: Optional[str] = ..., key: Optional[str] = ..., index: Optional[bool] = ..., info: Mapping[Any, Any] = ..., nullable: bool = ..., onupdate: Optional[Any] = ..., primary_key: bool = ..., server_default: Union[FetchedValue, str, TextClause, None] = ..., server_onupdate: Optional[FetchedValue] = ..., quote: Optional[bool] = ..., unique: Optional[bool] = ..., system: bool = ..., comment: Optional[str] = ..., **kwargs: Any) -> Column[_TE]
    note:     <1 more non-matching overload not shown>
    All other columns I have with datetime works, I suspect it's due to server_default
    Maico Timmerman
    @MaicoTimmerman
    That is a known issue: sqlalchemy/sqlalchemy2-stubs#81
    Mehdi ABAAKOUK
    @sileht
    great!
    Thank for your time, I will open a bug for the enum one
    Mehdi ABAAKOUK
    @sileht
    While I was writing the MCVE, I see it only occur if the Enum class is not in the same file.