Flask-SQLAlchemy check if row exists in table
I have a Flask application which uses Flask-SQLAlchemy to connect to a MySQL database. I would like to be able to check whether a row is present in a table. How would I modify a query like so to check the row exists:
db.session.query(User).filter_by(name='John Smith') I found a solution on this question which uses SQLAlchemy but does not seem to fit with the way Flask-SQLAlchemy works:
from sqlalchemy.sql import exists print session.query(exists().where(User.email == '. ')).scalar() 4 Answers 4
Since you only want to see if the user exists, you don’t want to query the entire object. Only query the id, it exists if the scalar return is not None.
exists = db.session.query(User.id).filter_by(name='davidism').first() is not None SELECT user.id AS user_id FROM user WHERE user.name = ? If you know name (or whatever field you’re querying) is unique, you can use scalar instead of first .
The second query you showed also works fine, Flask-SQLAlchemy does nothing to prevent any type of query that SQLAlchemy can make. This returns False or True instead of None or an id like above, but it is slightly more expensive because it uses a subquery.
exists = db.session.query(db.exists().where(User.name == 'davidism')).scalar() SELECT EXISTS (SELECT * FROM user WHERE user.name = ?) AS anon_1 Database helpers¶
Performs backend-specific testing to quickly determine if a database exists on the server.
database_exists('postgresql://postgres@localhost/name') #=> False create_database('postgresql://postgres@localhost/name') database_exists('postgresql://postgres@localhost/name') #=> True
Supports checking against a constructed URL as well.
engine = create_engine('postgresql://postgres@localhost/name') database_exists(engine.url) #=> False create_database(engine.url) database_exists(engine.url) #=> True
create_database¶
Issue the appropriate CREATE DATABASE statement.
- url – A SQLAlchemy engine URL.
- encoding – The encoding to create the database as.
- template – The name of the template from which to create the new database. At the moment only supported by PostgreSQL driver.
To create a database, you can pass a simple URL that would have been passed to create_engine .
create_database('postgresql://postgres@localhost/name')
You may also pass the url from an existing engine.
Has full support for mysql, postgres, and sqlite. In theory, other database engines should be supported.
drop_database¶
Issue the appropriate DROP DATABASE statement.
| Parameters: | url – A SQLAlchemy engine URL. |
|---|
Works similar to the create_database method in that both url text and a constructed url are accepted.
drop_database('postgresql://postgres@localhost/name') drop_database(engine.url)
has_index¶
Return whether or not given column or the columns of given foreign key constraint have an index. A column has an index if it has a single column index or it is the first column in compound column index.
A foreign key constraint has an index if the constraint columns are the first columns in compound column index.
| Parameters: | column_or_constraint – SQLAlchemy Column object or SA ForeignKeyConstraint object |
|---|
from sqlalchemy_utils import has_index class Article(Base): __tablename__ = 'article' id = sa.Column(sa.Integer, primary_key=True) title = sa.Column(sa.String(100)) is_published = sa.Column(sa.Boolean, index=True) is_deleted = sa.Column(sa.Boolean) is_archived = sa.Column(sa.Boolean) __table_args__ = ( sa.Index('my_index', is_deleted, is_archived), ) table = Article.__table__ has_index(table.c.is_published) # True has_index(table.c.is_deleted) # True has_index(table.c.is_archived) # False
Also supports primary key indexes
from sqlalchemy_utils import has_index class ArticleTranslation(Base): __tablename__ = 'article_translation' id = sa.Column(sa.Integer, primary_key=True) locale = sa.Column(sa.String(10), primary_key=True) title = sa.Column(sa.String(100)) table = ArticleTranslation.__table__ has_index(table.c.locale) # False has_index(table.c.id) # True
This function supports foreign key constraints as well
class User(Base): __tablename__ = 'user' first_name = sa.Column(sa.Unicode(255), primary_key=True) last_name = sa.Column(sa.Unicode(255), primary_key=True) class Article(Base): __tablename__ = 'article' id = sa.Column(sa.Integer, primary_key=True) author_first_name = sa.Column(sa.Unicode(255)) author_last_name = sa.Column(sa.Unicode(255)) __table_args__ = ( sa.ForeignKeyConstraint( [author_first_name, author_last_name], [User.first_name, User.last_name] ), sa.Index( 'my_index', author_first_name, author_last_name ) ) table = Article.__table__ constraint = list(table.foreign_keys)[0].constraint has_index(constraint) # True
has_unique_index¶
Return whether or not given column or given foreign key constraint has a unique index.
A column has a unique index if it has a single column primary key index or it has a single column UniqueConstraint.
A foreign key constraint has a unique index if the columns of the constraint are the same as the columns of table primary key or the coluns of any unique index or any unique constraint of the given table.
| Parameters: | column – SQLAlchemy Column object |
|---|
from sqlalchemy_utils import has_unique_index class Article(Base): __tablename__ = 'article' id = sa.Column(sa.Integer, primary_key=True) title = sa.Column(sa.String(100)) is_published = sa.Column(sa.Boolean, unique=True) is_deleted = sa.Column(sa.Boolean) is_archived = sa.Column(sa.Boolean) table = Article.__table__ has_unique_index(table.c.is_published) # True has_unique_index(table.c.is_deleted) # False has_unique_index(table.c.id) # True
This function supports foreign key constraints as well
class User(Base): __tablename__ = 'user' first_name = sa.Column(sa.Unicode(255), primary_key=True) last_name = sa.Column(sa.Unicode(255), primary_key=True) class Article(Base): __tablename__ = 'article' id = sa.Column(sa.Integer, primary_key=True) author_first_name = sa.Column(sa.Unicode(255)) author_last_name = sa.Column(sa.Unicode(255)) __table_args__ = ( sa.ForeignKeyConstraint( [author_first_name, author_last_name], [User.first_name, User.last_name] ), sa.Index( 'my_index', author_first_name, author_last_name, unique=True ) ) table = Article.__table__ constraint = list(table.foreign_keys)[0].constraint has_unique_index(constraint) # True
json_sql¶
Convert python data structures to PostgreSQL specific SQLAlchemy JSON constructs. This function is extremly useful if you need to build PostgreSQL JSON on python side.
This function needs PostgreSQL >= 9.4
Scalars are converted to to_json SQLAlchemy function objects
json_sql(1) # Equals SQL: to_json(1) json_sql('a') # to_json('a')
Mappings are converted to json_build_object constructs
json_sql('a': 'c', '2': 5>) # json_build_object('a', 'c', '2', 5)
Sequences (other than strings) are converted to json_build_array constructs
json_sql([1, 2, 3]) # json_build_array(1, 2, 3)
You can also nest these data structures
json_sql('a': [1, 2, 3]>) # json_build_object('a', json_build_array[1, 2, 3])
render_expression¶
Generate a SQL expression from the passed python expression.
Only the global variable, engine , is available for use in the expression. Additional local variables may be passed in the context parameter.
Note this function is meant for convenience and protected usage. Do NOT blindly pass user input to this function as it uses exec.
render_statement¶
Generate an SQL expression string with bound parameters rendered inline for the given SQLAlchemy statement.
© Copyright 2013-2022, Konsta Vesterinen Revision db32722a .
Versions latest stable Downloads pdf html epub On Read the Docs Project Home Builds Free document hosting provided by Read the Docs.
Как искать существование базы данных с sqlalchemy
Мне нужно написать скрипт с python sqlalchemy, который ищет, существует ли база данных, если база данных существует, он должен запросить базу данных, иначе создать базу данных и таблицы.
if db exists cursor.execute(sql) else create db test; create tables; insert data; 8 ответов
Вы можете использовать sqlalchemy.engine.base.Engine.connect() метод, который поднимет OperationalError если база данных не существует
import sqlalchemy as sqla db = sqla.create_engine(database_uri) try: db.connect() db.execute(sql) except OperationalError: # Switch database component of the uri default_database_uri = os.path.join(os.path.dirname( str(db.engine.url)), 'mysql') db = sqla.create_engine(default_database_uri) # Create your missing database/tables/data here # . Другой способ, если вы не против импортировать другие библиотеки, — это использовать sqlalchemy_utils. затем database_exists делает то, что вы ожидаете, что вы можете передать базу данных sqlalchemy URI.
if database_exists('sqllite:////tmp/test.db'): do_stuff_with_db(db) Я не знаю, каков канонический способ, но вот способ проверить, существует ли база данных, сверяясь со списком баз данных.
from sqlalchemy import create_engine # This engine just used to query for list of databases mysql_engine = create_engine('mysql://:@:'.format(user, pass, host, port)) # Query for existing databases existing_databases = mysql_engine.execute("SHOW DATABASES;") # Results are a list of single item tuples, so unpack each tuple existing_databases = [d[0] for d in existing_databases] # Create database if not exists if database not in existing_databases: mysql_engine.execute("CREATE DATABASE ".format(database)) print("Created database ".format(database)) # Go ahead and use this engine db_engine = create_engine('mysql://:@:/'.format(user, pass, host, port, db)) Вот альтернативный метод, если вам не нужно знать, была ли база данных создана или нет.
from sqlalchemy import create_engine # This engine just used to query for list of databases mysql_engine = create_engine('mysql://:@:'.format(user, pass, host, port)) # Query for existing databases mysql_engine.execute("CREATE DATABASE IF NOT EXISTS ".format(database)) # Go ahead and use this engine db_engine = create_engine('mysql://:@:/'.format(user, pass, host, port, db))