How can I dynamically change sqlalchemy __tablename__ at run time for purpose of localization using fastapi?

194 Views Asked by At

I am building a multilingual api for an educational app using fastapi.
I have created differrent tables for translated parts and created mysql views for differrent languages.
I did not want to write differrent pydantic classes for each translated table parts.

Instead I wrote sone functions to manipulate sqlalchemy _tablename_ attribute of sqlachemy tables. This way I thought I could reach differrent tables like lessons_en or lessons_de which are just mysql views for translated parts. Table structure and view structure matches the original sqlalchemy model and pydantic schema.

e.g.

def get_table_suffix():
    lang = sys_vars.CURRENT_LOCAL__LANG
    print(str(lang) + "from get_table_suffix")
    match lang:
        case "":
            return ""
        case "en":
            return "_en"
        case "de":
            return "_de"


def get_localized_table_name(tbl_nm: str):
    table_name = tbl_nm + get_table_suffix()
    return table_name

then..

class LessonSummary(Base):
    __tablename__ = get_localized_table_name("lessons")

for matching correct language view then at login function I call ..

    settings: dict = get_user_settings(user)
    set_user_lang(settings)
    sys_vars.CURRENT_LOCAL__LANG = sys_vars.LOCAL_LANGUAGES[settings["language"]]

these functions set CURRENT_LOCAL__LANG which happens ( I tested it ) But since the tables are loaded, changing tablename does not change tables.

def get_user_settings(user: schemas.User):
    settings: dict = {}
    for setting in user.Settings:
        settings[setting.SettingName] = setting.SettingValue
    return settings


def set_user_lang(settings: dict):
    sys_vars.CURRENT_LOCAL__LANG = sys_vars.LOCAL_LANGUAGES[settings["language"]]
    print(sys_vars.CURRENT_LOCAL__LANG + " from set_user_lang")
    return sys_vars.CURRENT_LOCAL__LANG

However, _tablename_ attribute is set at loading time and I cannot seem to change it at run time. I tried to do it before the app loaded routers but user login happens after app startup and I need to reach the instance of the table. I tried MetaData() obj but no-luck

So, my question.. Am I going the wrong way about this problem? Can I dynamically change the _tablename_ at run time and reflect the changes in session. Do I need a reload somehow? I did manage to store language correctly inside jwt but I can't change the tablename. So How can I change it at runtime?

I tried changing the tablename dynamically but sine the app had already loaded I was not successful.

2

There are 2 best solutions below

0
Yurii Motov On

You can use imperative mapping style instead of declarative style. With imperative style your model classes are not linked to metadata until you link them by calling mapper_registry.map_imperatively().

But in this case there are some duplication (you have to declare fields twice - in model and in table).

from dataclasses import dataclass, field

from sqlalchemy.orm import registry, sessionmaker
from sqlalchemy import create_engine
from sqlalchemy import MetaData, Table, Column, Integer, String

mapper_registry = registry()
metadata = MetaData()

user_lang: str = ''

@dataclass
class LessonSummary:
    id: int = field(init=False)
    title: str


def get_localized_table_name(tb_name):
    return f"{tb_name}_{user_lang}"


def create_localized_tables(engine):
    localized_table_name = get_localized_table_name("lessons")

    lesson_summary_table = Table(
        localized_table_name,
        mapper_registry.metadata,
        Column('id', Integer, primary_key=True),
        Column('title', String(50))
    )

    mapper_registry.map_imperatively(LessonSummary, lesson_summary_table)

    mapper_registry.metadata.create_all(bind=engine)


engine = create_engine('sqlite://', echo=True)

user_lang = input("Language:")

# When you have all necessary data, create tables
create_localized_tables(engine)

# And you can use your models
session_maker = sessionmaker(bind=engine)

with session_maker() as session:

    l1 = LessonSummary("Lesson 1")

    session.add(l1)
    session.commit()

    lesson = session.get(LessonSummary, 1)
    print(lesson.title)

Output:

CREATE TABLE lessons_me (
        id INTEGER NOT NULL, 
        title VARCHAR(50), 
        PRIMARY KEY (id)
)

INSERT INTO lessons_me (title) VALUES (?)
('Lesson 1',)

FROM lessons_me 
WHERE lessons_me.id = ?
(1,)

Lesson 1
0
yavuzakyazici On

I am answering my own question since I don't think there is an answer to my original question which was dynamically changing sqlalchemy tables pointing at differrent tables or views depending on user's local language without duplicating a lot code.

My database design was right for future exandibility.

I had a table for translation languages and a lessons_translation table

lessons
-------
id
title
description
body

lessons_trs
----------
id
lesson_id
language_id
title
description
body

I also had a view with join of lessons and lessons_trs. let's say for german selected view

select lessons.id, lessons_trs.title, lessons_trs.description, lessons_trs.body 
from lessons_trs, lessons
where lessons_trs.language_id="de" 
and lessons_trs.lesson_id=lessons.id

and the view with select statement above was called lessons_de. This gives me a translated table view like..

lessons_de
----------
id
title
description
body

..which matches pydantic schemas since lessons and lessons_de data structure match exactly.

It would be as simple as changing sqlalchemy _tablename_ and refreshing the session to point to another table at runtime.

Even though Yurii's solution is a working one, it requires me to duplicate data classes as many as translated language.

So I ended up creating new Tables inside sqlalchemy models and adding a language checking if statement for all my crud functions. It's still better than adding matching pydantic classes for each translation.

I am a bit disappointed since I was expecting better localization options from sqlalchemy and fastapi.

May be they have a mechanism for it and I could not find it yet but I have to go on since I've already spent a few days on this subject.

I wanted to leave the answer here for possible future solutions or updates.