Is it possible to use SQLAlchemy mappers (or similar)?

[petedmarsh]

Is there a way to automatically map a result row from a query to a class, something like SQLAlchemy's mappers, e.g.:

notes = sqlalchemy.Table(
    "notes",
    metadata,
    sqlalchemy.Column("id", sqlalchemy.Integer, primary_key=True),
    sqlalchemy.Column("text", sqlalchemy.String(length=100)),
    sqlalchemy.Column("completed", sqlalchemy.Boolean),
)

class Note:
    def __init__(self, id, ...):
         self.id = id
         # ... 

rows = await databse.fetch_all(select([notes]))

note_objects = ... # ???

note_objects == [Note(1, ...), Note(2, ...)]

For simple cases like the above I can do the mapping myself fairly easily but for complicated joins the code becomes ugly and repetitive.

For clarity, I don't want/or need a full ORM, I just want to map the database rows to some data classes for ease of use elsewhere.

[artslob]

Try pydantic. Parsing list of db rows would be look like this:

from typing import List

from pydantic import BaseModel


class Note(BaseModel):
    a: int
    b: str


# rows from database provide dict-like interface
x: List[dict] = [dict(a=1, b="2"), dict(a=3, b="4")]

y: List[Note] = [Note.parse_obj(i) for i in x]

print(y)

Output:

[Note(a=1, b='2'), Note(a=3, b='4')]

Pydantic also allows to parse complicated and nested dictionaries to objects and supports type hinting (if you use IDE, eg PyCharm).

[ikhomutov]

I can share my workaround for this. Hope this will be helpful.

notes = sqlalchemy.Table(
    "notes",
    metadata,
    sqlalchemy.Column("id", sqlalchemy.Integer, primary_key=True),
    sqlalchemy.Column("text", sqlalchemy.String(length=100)),
    sqlalchemy.Column("completed", sqlalchemy.Boolean),
)

from dataclasses import dataclass, fields

@dataclass
class DBModel:
    __table__ = None

    @classmethod
    def from_db(cls, record):
        d = {}
        for field in fields(cls):
            d.update({field.name: record[cls.__table__.c[field.name]]})
        # If we use outerjoin and corresponding record not exist
        if d['id'] is None:
            return None
        return cls(**d)

@dataclass
class Note:
    id: int
    text: str
    completed: bool

    __table__ = notes

Now you can convert your Record object into Note object

rows = await databse.fetch_all(select([notes]))
for row in rows:
    note = Note.from_db(row)

[graingert]

sqlalchemy 1.4 supports the full ORM when used with asyncio: https://docs.sqlalchemy.org/en/14/orm/extensions/asyncio.html#synopsis-orm