This is an old question, but still relevant today. Getting SQL Alchemy to return a dictionary is very useful, especially when working with RESTful based APIs that return JSON. Here is how I did it using the db_session in Python 3: resultproxy = db_session.execute(query) d, a = {}, [] for rowproxy in resultproxy: # rowproxy.items() … Read more
From version 1.2 SQLAlchemy will support on_duplicate_key_update for MySQL There is also examples of how to use it: from sqlalchemy.dialects.mysql import insert insert_stmt = insert(my_table).values( id=’some_existing_id’, data=”inserted value”) on_duplicate_key_stmt = insert_stmt.on_duplicate_key_update( data=insert_stmt.values.data, status=”U” ) conn.execute(on_duplicate_key_stmt) From version 1.1 SQLAlchemy support on_conflict_do_update for PostgreSQL Examples: from sqlalchemy.dialects.postgresql import insert insert_stmt = insert(my_table).values( id=’some_existing_id’, data=”inserted value”) do_update_stmt … Read more
session.query is the old API. The asynchronous version uses select and accompanying methods. from sqlalchemy.future import select from sqlalchemy.ext.asyncio import AsyncSession, create_async_engine from sqlalchemy.orm import sessionmaker engine = create_async_engine(_build_async_db_uri(CONFIG.general.sqlalchemy_db_uri)) async_session = sessionmaker( engine, expire_on_commit=False, class_=AsyncSession ) CACHE = {} async def _load_all(): global CACHE try: async with async_session() as session: q = select(TableClass) result = … Read more
You need to perform initialization after installation: $ export AIRFLOW_HOME=some/dir $ airflow db init # or `airflow initdb` for the legacy 1.X If AIRFLOW_HOME is unset, ~/airflow/ will be created and used. This is where the config and logs will be stored; if you want to reset the configuration, remove the dir stored in AIRFLOW_HOME … Read more
I had the same question and the answer, which I found, is: if not engine.dialect.has_schema(engine, schema_name): engine.execute(sqlalchemy.schema.CreateSchema(schema_name)) We can also check schema without engine instance, but using connection conn = engine.connect() if conn.dialect.has_schema(conn, schema_name): …
Finally got answer from Flask documentation, we can get cursor from db object using, from sqlalchemy import create_engine engine = create_engine(‘your_connection_string’) connection = engine.raw_connection() cursor = connection.cursor()
Those two lines are the same thing. Only exceptions raised differ. In fact, get() is implemented on top of one(). There would be a difference if your filter() returned more than a result, but this is indeed not possible in your case. By the way, SQL does not have a GET operation, it only has … Read more
The doc you reference explains the issue: This parameter is only used when issuing CREATE TABLE statements. If you originally created your database without nullable=False (or created it in some other way, separate from SQLAlchemy, without NOT NULL on that column), then your database column doesn’t have that constraint information. This information lives in reference … Read more
Figured it out: connection_string: str = “sqlite:///:memory:” engine = create_engine(connection_string) session = Session(bind=engine) print(type(engine)) # sqlalchemy.engine.base.Engine print(type(session)) # sqlalchemy.orm.session.Session Thus, type hinting is achieved the following way for example: from sqlalchemy.engine.base import Engine def test_func(engine: Engine): pass
Edit If you’re experiencing this, upgrading Flask-SQLAlchemy to >= 2.5 should resolve the issue per https://github.com/pallets/flask-sqlalchemy/issues/910#issuecomment-802098285. Pinning SQLAlchemy to ~1.3 should no longer be necessary. I ran into this issue a little earlier, but think I’ve figured out what’s going on. SQLAlchemy is automatically installed as a dependency for Flask-SQLAlchemy and its latest release (1.4.0) … Read more