Sqlalchemy обработка ошибок - Не ошибается лишь тот, кто ничего не делает!

How do you handle errors in SQLAlchemy? I am relatively new to SQLAlchemy and do not know yet.

Before I used SQLAlchemy, I would do things like

status = db.query("INSERT INTO users ...")
if (!status):
    raise Error, db.error

But now I am coding in SQLAlchemy and I do things like

user = User('Boda Cydo')
session.add(user)
session.commit()

No error checking whatsoever!

I do not like this coding style without error checking at all.

Please advice on how to check and handle errors in SQLAlchemy!

Sincerely, Boda Cydo.

asked Jan 26, 2010 at 0:21

Your example says:

status = db.query("INSERT INTO users ...")
if (!status):
    raise Error, db.error

That seems to mean that you want to raise an exception if there’s some error on the query (with raise Error, db.error). However sqlalchemy already does that for you — so

user = User('Boda Cydo')
session.add(user)
session.commit()

Is just the same. The check-and-raise part is already inside SQLAlchemy.

Here is a list of the errors sqlalchemy itself can raise, taken from help(sqlalchemy.exc) and help(sqlalchemy.orm.exc):

sqlalchemy.exc:
- ArgumentError — Raised when an invalid or conflicting function argument is supplied.
  This error generally corresponds to construction time state errors.
- CircularDependencyError — Raised by topological sorts when a circular dependency is detected
- CompileError — Raised when an error occurs during SQL compilation
- ConcurrentModificationError
- DBAPIError — Raised when the execution of a database operation fails.
  If the error-raising operation occured in the execution of a SQL
  statement, that statement and its parameters will be available on
  the exception object in the statement and params attributes.
  The wrapped exception object is available in the orig attribute.
  Its type and properties are DB-API implementation specific.
- DataError Wraps a DB-API DataError.
- DatabaseError — Wraps a DB-API DatabaseError.
- DisconnectionError — A disconnect is detected on a raw DB-API connection.
  be raised by a PoolListener so that the host pool forces a disconnect.
- FlushError
- IdentifierError — Raised when a schema name is beyond the max character limit
- IntegrityError — Wraps a DB-API IntegrityError.
- InterfaceError — Wraps a DB-API InterfaceError.
- InternalError — Wraps a DB-API InternalError.
- InvalidRequestError — SQLAlchemy was asked to do something it can’t do. This error generally corresponds to runtime state errors.
- NoReferenceError — Raised by ForeignKey to indicate a reference cannot be resolved.
- NoReferencedColumnError — Raised by ForeignKey when the referred Column cannot be located.
- NoReferencedTableError — Raised by ForeignKey when the referred Table cannot be located.
- NoSuchColumnError — A nonexistent column is requested from a RowProxy.
- NoSuchTableError — Table does not exist or is not visible to a connection.
- NotSupportedError — Wraps a DB-API NotSupportedError.
- OperationalError — Wraps a DB-API OperationalError.
- ProgrammingError — Wraps a DB-API ProgrammingError.
- SADeprecationWarning — Issued once per usage of a deprecated API.
- SAPendingDeprecationWarning — Issued once per usage of a deprecated API.
- SAWarning — Issued at runtime.
- SQLAlchemyError — Generic error class.
- SQLError — Raised when the execution of a database operation fails.
- TimeoutError — Raised when a connection pool times out on getting a connection.
- UnboundExecutionError — SQL was attempted without a database connection to execute it on.
- UnmappedColumnError
sqlalchemy.orm.exc:
- ConcurrentModificationError — Rows have been modified outside of the unit of work.
- FlushError — A invalid condition was detected during flush().
- MultipleResultsFound — A single database result was required but more than one were found.
- NoResultFound — A database result was required but none was found.
- ObjectDeletedError — A refresh() operation failed to re-retrieve an object’s row.
- UnmappedClassError — A mapping operation was requested for an unknown class.
- UnmappedColumnError — Mapping operation was requested on an unknown column.
- UnmappedError — TODO
- UnmappedInstanceError — A mapping operation was requested for an unknown instance.

MarredCheese

17k8 gold badges91 silver badges89 bronze badges

answered Jan 26, 2010 at 3:37

nosklonosklo

216k56 gold badges293 silver badges296 bronze badges

I tried this and it showed me the specific error message.

from sqlalchemy.exc import SQLAlchemyError

try:
# try something

except SQLAlchemyError as e:
  error = str(e.__dict__['orig'])
  return error

Hope this helps

answered Dec 8, 2018 at 13:10

ToufiqToufiq

1,07612 silver badges16 bronze badges

To get type of exception, you can simply call :

Firstly, import exc from sqlalchemy

from sqlalchemy import exc

To catch types of errors:

    try:
        # any query
    except exc.SQLAlchemyError as e:
        print(type(e))

This will give you exception type:
Output:

<class 'sqlalchemy.exc.IntegrityError'>
<class 'sqlalchemy.exc.IntegrityError'>

answered Apr 25, 2020 at 23:06

MoidMoid

3803 silver badges9 bronze badges

My two cents on handling errors in SQLAlchemy: a simple python’s try-except will not work as MySQL is persistent. For example, if you try to insert a record to the database but it is a duplicate, the program will take the exception route but MySQL will stop based on the insert command that did not go through. Avoid try-except in combination with SQLAlchemy commands or be prepared for these situations.

answered Mar 10 at 11:22

Al MartinsAl Martins

4135 silver badges13 bronze badges

SQLAlchemy will raise an exception on error….

answered Jan 26, 2010 at 0:24

tholotholo

1572 bronze badges

Источник

As far as I know, the only way is to parse the error string in the exception. I can’t find any tuple specifying the error, the column with the violated uniqueness constraint and value separately.

Specifically, one can search the exception message using a substring search operator or a regular expression. For example:

    db.session.add(SupplierUser(supplier_id=supplier_id, username=username, password=password, creator=user))
    try:
        db.session.commit()
    except IntegrityError as err:
        db.session.rollback()
        if "UNIQUE constraint failed: user.username" in str(err):
            return False, "error, username already exists (%s)" % username
        elif "FOREIGN KEY constraint failed" in str(err):
            return False, "supplier does not exist"
        else:
            return False, "unknown error adding user"

However, these strings can be quite long because the SQL statement is included, for example:

(sqlite3.IntegrityError) UNIQUE constraint failed: user.username [SQL: 'INSERT INTO user (username, password_hash, created_time, creator_id, role, is_active, supplier_id) VALUES (?, ?, ?, ?, ?, ?, ?)'] [parameters: ('bob', ...

Thus, you will minimise the error handling latencies parsing exceptions if you search through the database error message, without the added information from sqlalchemy. This can be done by examining err.args, which should be smaller:

'(sqlite3.IntegrityError) UNIQUE constraint failed: supplier_user.username',)

The updated example:

    db.session.add(SupplierUser(supplier_id=supplier_id, username=username, password=password, creator=user))
    try:
        db.session.commit()
    except IntegrityError as err:
        db.session.rollback()
        err_msg = err.args[0]
        if "UNIQUE constraint failed: supplier_user.username" in err_msg:
            return False, "error, supplier username already exists (%s)" % username
        elif "FOREIGN KEY constraint failed" in err_msg:
            return False, "supplier does not exist"
        else:
            return False, "unknown error adding user"

Note the error syntax I used here is for sqlite3. Parsing an mysql exception error message like:

(1062, "Duplicate entry 'usr544' for key 'username'")

Can be done with an appropriate regular expression. Note that it looks like a tuple, but it is actually a string (sqlalchemy version 1.1.3 and mysql 5.5).

For example:

    except IntegrityError as err:
        db.session.rollback()
        if re.match("(.*)Duplicate entry(.*)for key 'username'(.*)", err.args[0]):
    .... etc ....

Источник

Exceptions used with SQLAlchemy.

The base exception class is SQLAlchemyError. Exceptions which are
raised as a result of DBAPI exceptions are all subclasses of
DBAPIError.

exception sqlalchemy.exc.AmbiguousForeignKeysError¶: Raised when more than one foreign key matching can be located
between two selectables during a join.

exception sqlalchemy.exc.ArgumentError¶

Raised when an invalid or conflicting function argument is supplied.

This error generally corresponds to construction time state errors.

exception sqlalchemy.exc.AwaitRequired¶: Error raised by the async greenlet spawn if no async operation
was awaited when it required one.

exception sqlalchemy.exc.Base20DeprecationWarning¶

Issued for usage of APIs specifically deprecated or legacy in
SQLAlchemy 2.0.

attribute sqlalchemy.exc.Base20DeprecationWarning.deprecated_since: str | None = ‘1.4’¶: Indicates the version that started raising this deprecation warning

exception sqlalchemy.exc.CircularDependencyError¶

Raised by topological sorts when a circular dependency is detected.

There are two scenarios where this error occurs:

In a Session flush operation, if two objects are mutually dependent
on each other, they can not be inserted or deleted via INSERT or
DELETE statements alone; an UPDATE will be needed to post-associate
or pre-deassociate one of the foreign key constrained values.
The post_update flag described at Rows that point to themselves / Mutually Dependent Rows can resolve
this cycle.
In a MetaData.sorted_tables operation, two
ForeignKey
or ForeignKeyConstraint objects mutually refer to each
other. Apply the use_alter=True flag to one or both,
see Creating/Dropping Foreign Key Constraints via ALTER.

method sqlalchemy.exc.CircularDependencyError.__init__(message: str, cycles: Any, edges: Any, msg: str | None = None, code: str | None = None)¶

exception sqlalchemy.exc.CompileError¶: Raised when an error occurs during SQL compilation

exception sqlalchemy.exc.ConstraintColumnNotFoundError¶

raised when a constraint refers to a string column name that
is not present in the table being constrained.

New in version 2.0.

exception sqlalchemy.exc.DBAPIError¶

Raised when the execution of a database operation fails.

Wraps exceptions raised by the DB-API underlying the
database operation. Driver-specific implementations of the standard
DB-API exception types are wrapped by matching sub-types of SQLAlchemy’s
DBAPIError when possible. DB-API’s Error type maps to
DBAPIError in SQLAlchemy, otherwise the names are identical. Note
that there is no guarantee that different DB-API implementations will
raise the same exception type for any given error condition.

DBAPIError features StatementError.statement
and StatementError.params attributes which supply context
regarding the specifics of the statement which had an issue, for the
typical case when the error was raised within the context of
emitting a SQL statement.

The wrapped exception object is available in the
StatementError.orig attribute. Its type and properties are
DB-API implementation specific.

method sqlalchemy.exc.DBAPIError.__init__(statement: str | None, params: _AnyExecuteParams | None, orig: BaseException, hide_parameters: bool = False, connection_invalidated: bool = False, code: str | None = None, ismulti: bool | None = None)¶

exception sqlalchemy.exc.DataError¶: Wraps a DB-API DataError.

exception sqlalchemy.exc.DatabaseError¶: Wraps a DB-API DatabaseError.

exception sqlalchemy.exc.DisconnectionError¶

A disconnect is detected on a raw DB-API connection.

This error is raised and consumed internally by a connection pool. It can
be raised by the PoolEvents.checkout()
event so that the host pool
forces a retry; the exception will be caught three times in a row before
the pool gives up and raises InvalidRequestError
regarding the connection attempt.

Object Name	Description
DontWrapMixin	A mixin class which, when applied to a user-defined Exception class, will not be wrapped inside of `StatementError` if the error is emitted within the process of executing a statement.
HasDescriptionCode	helper which adds ‘code’ as an attribute and ‘_code_str’ as a method

class sqlalchemy.exc.DontWrapMixin¶

A mixin class which, when applied to a user-defined Exception class,
will not be wrapped inside of StatementError if the error is
emitted within the process of executing a statement.

E.g.:

from sqlalchemy.exc import DontWrapMixin

class MyCustomException(Exception, DontWrapMixin):
    pass

class MySpecialType(TypeDecorator):
    impl = String

    def process_bind_param(self, value, dialect):
        if value == 'invalid':
            raise MyCustomException("invalid!")

exception sqlalchemy.exc.DuplicateColumnError¶

a Column is being added to a Table that would replace another
Column, without appropriate parameters to allow this in place.

New in version 2.0.0b4.

class sqlalchemy.exc.HasDescriptionCode¶: helper which adds ‘code’ as an attribute and ‘_code_str’ as a method

exception sqlalchemy.exc.IdentifierError¶: Raised when a schema name is beyond the max character limit

exception sqlalchemy.exc.IllegalStateChangeError¶

An object that tracks state encountered an illegal state change
of some kind.

New in version 2.0.

exception sqlalchemy.exc.IntegrityError¶: Wraps a DB-API IntegrityError.

exception sqlalchemy.exc.InterfaceError¶: Wraps a DB-API InterfaceError.

exception sqlalchemy.exc.InternalError¶: Wraps a DB-API InternalError.

exception sqlalchemy.exc.InvalidRequestError¶

SQLAlchemy was asked to do something it can’t do.

This error generally corresponds to runtime state errors.

exception sqlalchemy.exc.InvalidatePoolError¶

Raised when the connection pool should invalidate all stale connections.

A subclass of DisconnectionError that indicates that the
disconnect situation encountered on the connection probably means the
entire pool should be invalidated, as the database has been restarted.

This exception will be handled otherwise the same way as
DisconnectionError, allowing three attempts to reconnect
before giving up.

New in version 1.2.

exception sqlalchemy.exc.LegacyAPIWarning¶: indicates an API that is in ‘legacy’ status, a long term deprecation.

exception sqlalchemy.exc.MissingGreenlet¶: Error raised by the async greenlet await_ if called while not inside
the greenlet spawn context.

exception sqlalchemy.exc.MovedIn20Warning¶: Subtype of RemovedIn20Warning to indicate an API that moved only.

exception sqlalchemy.exc.MultipleResultsFound¶

A single database result was required but more than one were found.

Changed in version 1.4: This exception is now part of the
sqlalchemy.exc module in Core, moved from the ORM. The symbol
remains importable from sqlalchemy.orm.exc.

exception sqlalchemy.exc.NoForeignKeysError¶: Raised when no foreign keys can be located between two selectables
during a join.

exception sqlalchemy.exc.NoInspectionAvailable¶: A subject passed to sqlalchemy.inspection.inspect() produced
no context for inspection.

exception sqlalchemy.exc.NoReferenceError¶: Raised by ForeignKey to indicate a reference cannot be resolved.

exception sqlalchemy.exc.NoReferencedColumnError¶

Raised by ForeignKey when the referred Column cannot be
located.

method sqlalchemy.exc.NoReferencedColumnError.__init__(message: str, tname: str, cname: str)¶

exception sqlalchemy.exc.NoReferencedTableError¶

Raised by ForeignKey when the referred Table cannot be
located.

method sqlalchemy.exc.NoReferencedTableError.__init__(message: str, tname: str)¶

exception sqlalchemy.exc.NoResultFound¶

A database result was required but none was found.

Changed in version 1.4: This exception is now part of the
sqlalchemy.exc module in Core, moved from the ORM. The symbol
remains importable from sqlalchemy.orm.exc.

exception sqlalchemy.exc.NoSuchColumnError¶: A nonexistent column is requested from a Row.

Class signature

class sqlalchemy.exc.NoSuchColumnError (sqlalchemy.exc.InvalidRequestError, builtins.KeyError)

exception sqlalchemy.exc.NoSuchModuleError¶: Raised when a dynamically-loaded module (usually a database dialect)
of a particular name cannot be located.

exception sqlalchemy.exc.NoSuchTableError¶: Table does not exist or is not visible to a connection.

exception sqlalchemy.exc.NotSupportedError¶: Wraps a DB-API NotSupportedError.

exception sqlalchemy.exc.ObjectNotExecutableError¶

Raised when an object is passed to .execute() that can’t be
executed as SQL.

method sqlalchemy.exc.ObjectNotExecutableError.__init__(target: Any)¶

exception sqlalchemy.exc.OperationalError¶: Wraps a DB-API OperationalError.

exception sqlalchemy.exc.PendingRollbackError¶

A transaction has failed and needs to be rolled back before
continuing.

New in version 1.4.

exception sqlalchemy.exc.ProgrammingError¶: Wraps a DB-API ProgrammingError.

exception sqlalchemy.exc.ResourceClosedError¶: An operation was requested from a connection, cursor, or other
object that’s in a closed state.

exception sqlalchemy.exc.SADeprecationWarning¶

Issued for usage of deprecated APIs.

Class signature

class sqlalchemy.exc.SADeprecationWarning (sqlalchemy.exc.HasDescriptionCode, builtins.DeprecationWarning)

attribute sqlalchemy.exc.SADeprecationWarning.deprecated_since: str | None = None¶: Indicates the version that started raising this deprecation warning

exception sqlalchemy.exc.SAPendingDeprecationWarning¶

A similar warning as SADeprecationWarning, this warning
is not used in modern versions of SQLAlchemy.

Class signature

class sqlalchemy.exc.SAPendingDeprecationWarning (builtins.PendingDeprecationWarning)

attribute sqlalchemy.exc.SAPendingDeprecationWarning.deprecated_since: str | None = None¶: Indicates the version that started raising this deprecation warning

exception sqlalchemy.exc.SATestSuiteWarning¶

warning for a condition detected during tests that is non-fatal

Currently outside of SAWarning so that we can work around tools like
Alembic doing the wrong thing with warnings.

Class signature

class sqlalchemy.exc.SATestSuiteWarning (builtins.Warning)

exception sqlalchemy.exc.SAWarning¶: Issued at runtime.

Class signature

class sqlalchemy.exc.SAWarning (sqlalchemy.exc.HasDescriptionCode, builtins.RuntimeWarning)

exception sqlalchemy.exc.SQLAlchemyError¶: Generic error class.

Class signature

class sqlalchemy.exc.SQLAlchemyError (sqlalchemy.exc.HasDescriptionCode, builtins.Exception)

exception sqlalchemy.exc.StatementError¶

An error occurred during execution of a SQL statement.

StatementError wraps the exception raised
during execution, and features statement
and params attributes which supply context regarding
the specifics of the statement which had an issue.

The wrapped exception object is available in
the orig attribute.

method sqlalchemy.exc.StatementError.__init__(message: str, statement: str | None, params: _AnyExecuteParams | None, orig: BaseException | None, hide_parameters: bool = False, code: str | None = None, ismulti: bool | None = None)¶

attribute sqlalchemy.exc.StatementError.ismulti: bool | None = None¶: multi parameter passed to repr_params(). None is meaningful.

attribute sqlalchemy.exc.StatementError.orig: BaseException | None = None¶: The original exception that was thrown.

attribute sqlalchemy.exc.StatementError.params: _AnyExecuteParams | None = None¶: The parameter list being used when this exception occurred.

attribute sqlalchemy.exc.StatementError.statement: str | None = None¶: The string SQL statement being invoked when this exception occurred.

exception sqlalchemy.exc.TimeoutError¶: Raised when a connection pool times out on getting a connection.

exception sqlalchemy.exc.UnboundExecutionError¶: SQL was attempted without a database connection to execute it on.

exception sqlalchemy.exc.UnreflectableTableError¶

Table exists but can’t be reflected for some reason.

New in version 1.2.

exception sqlalchemy.exc.UnsupportedCompilationError¶

Raised when an operation is not supported by the given compiler.

method sqlalchemy.exc.UnsupportedCompilationError.__init__(compiler: Compiled | TypeCompiler, element_type: Type[ClauseElement], message: str | None = None)¶

Источник

Время на прочтение
14 мин

Количество просмотров 52K

blog.miguelgrinberg.com

Miguel Grinberg

<<< предыдущая следующая >>>

Эта статья является переводом седьмой части нового издания учебника Мигеля Гринберга, выпуск которого автор планирует завершить в мае 2018.Прежний перевод давно утратил свою актуальность.

Я, со своей стороны, постараюсь не отставать с переводом.

Это седьмая глава серии Flask Mega-Tutorial, в которой я расскажу вам, как выполнять обработку ошибок в приложении Flask.

Для справки ниже приведен список статей этой серии.

Примечание 1: Если вы ищете старые версии данного курса, это здесь.

Примечание 2: Если вдруг Вы хотели бы выступить в поддержку моей(Мигеля) работы в этом блоге, или просто не имеете терпения дожидаться неделю статьи, я (Мигель Гринберг)предлагаю полную версию данного руководства упакованную электронную книгу или видео. Для получения более подробной информации посетите learn.miguelgrinberg.com.

В этой главе я перехожу от кодирования новых функций для моего микроблогического приложения и вместо этого обсужу несколько стратегий борьбы с ошибками, которые неизменно появляются в любом программном проекте. Чтобы проиллюстрировать эту тему, я намеренно допустил ошибку в коде, который я добавил в главе 6. Прежде чем продолжить чтение, посмотрите, сможете ли вы его найти!

Ссылки GitHub для этой главы: Browse, Zip, Diff.

Обработка ошибок в Flask

Что происходит, когда возникает ошибка в приложении Flask? Лучший способ узнать это — испытать это самому. Запустите приложение и убедитесь, что у вас зарегистрировано не менее двух пользователей. Войдите в систему как один из пользователей, откройте страницу профиля и нажмите ссылку «Изменить». В редакторе профиля попробуйте изменить имя пользователя на существующее имя другого пользователя, который уже зарегистрирован, и попытайтесь применить исправления! Это приведет к появлению страшной страницы «Internal Server Error» ( «Внутренняя ошибка сервера» ):

В сеансе терминала, на котором запущено приложение, вы видите трассировку стека ошибки. Трассировки стека чрезвычайно полезны при отладке ошибок, поскольку они показывают последовательность вызовов в этом стеке, вплоть до строки, вызвавшей ошибку:

(venv) $ flask run
 * Serving Flask app "microblog"
 * Running on http://127.0.0.1:5000/ (Press CTRL+C to quit)
[2017-09-14 22:40:02,027] ERROR in app: Exception on /edit_profile [POST]
Traceback (most recent call last):
  File "/home/miguel/microblog/venv/lib/python3.6/site-packages/sqlalchemy/engine/base.py", line 1182, in _execute_context
    context)
  File "/home/miguel/microblog/venv/lib/python3.6/site-packages/sqlalchemy/engine/default.py", line 470, in do_execute
    cursor.execute(statement, parameters)
sqlite3.IntegrityError: UNIQUE constraint failed: user.username

Трассировка стека указывает, чем вызвана ошибка. Приложение позволяет пользователю изменять имя пользователя без проверки, что новое имя пользователя не совпадает с другим пользователем, уже находящимся в системе. Ошибка возникает из SQLAlchemy, которая пытается записать новое имя пользователя в базу данных, но база данных отвергает его, потому что столбец имени пользователя определен с unique = True.

Важно, что страница с ошибкой, представленная пользователю, не содержит много информации об ошибке, и это правильно. Я определенно не хочу, чтобы пользователи узнали, что авария была вызвана ошибкой базы данных или какой базой данных я пользуюсь, а также именами таблиц и полей в моей базе данных. Вся эта информация должна быть внутренней.

Есть несколько вещей, которые далеки от идеала. У меня есть страница с ошибкой, которая безобразна и не соответствует макету приложения. У меня также есть важные трассировки стека приложений, которые сбрасываются на терминале, и мне нужно постоянно следить за тем, чтобы я не пропустил никаких ошибок. И, конечно, у меня есть ошибка. Я собираюсь решить все эти проблемы, но сначала поговорим о режиме отладки Flask.

Режим отладки

То, как ошибки обрабатываются выше, отлично подходит для системы, которая работает на production сервере. Если есть ошибка, пользователь получает страницу с неопределенной ошибкой (хотя я собираюсь сделать эту страницу с ошибкой более приятной), а важные данные об ошибке — в выводе сервера или в файле журнала.

Но когда вы разрабатываете приложение, вы можете включить режим отладки, режим, в котором Flask выводит действительно хороший отладчик непосредственно в ваш браузер. Чтобы активировать режим отладки, остановите приложение, а затем установите следующую переменную среды:

(venv) $ export FLASK_DEBUG=1

Если вы работаете в ОС Microsoft Windows, не забудьте использовать set вместо экспорта.

После того, как вы установили FLASK_DEBUG, перезапустите сервер. Строки на вашем терминале будут немного отличаться от того, что вы привыкли видеть:

(venv) microblog2 $ flask run
 * Serving Flask app "microblog"
 * Forcing debug mode on
 * Running on http://127.0.0.1:5000/ (Press CTRL+C to quit)
 * Restarting with stat
 * Debugger is active!
 * Debugger PIN: 177-562-960

Теперь устроим приложению аварийный сбой еще раз, чтобы увидеть интерактивный отладчик в вашем браузере:

Отладчик позволяет развернуть каждый уровень стека и увидеть соответствующий исходный код. Вы также можете открыть Python для любого из фреймов и выполнить любые допустимые выражения Python, например, чтобы проверить значения переменных.

Крайне важно, чтобы вы никогда не запускали приложение Flask в режиме отладки на рабочем сервере. Отладчик позволяет удаленно выполнять код на сервере, поэтому он может стать неожиданным подарком злоумышленнику, который хочет проникнуть в ваше приложение или на ваш сервер. В качестве дополнительной меры безопасности отладчик, запущенный в браузере, закроется, и при первом использовании запросит PIN-код, который вы можете увидеть на выходе команды flask run.

Поскольку я говорю о режиме отладки, следует упомянуть про вторую важную функцию, которая включена в режиме отладки — перезагрузка. Это очень полезная функция разработки, которая автоматически перезапускает приложение при изменении исходного файла. Если вы выполните flask run в режиме отладки, можно продолжать работать в своем приложении и при каждом сохранении файла, приложение перезапустится, чтобы забрать новый код.

Пользовательские страницы ошибок

Flask предоставляет механизм приложения для создания собственных страниц ошибок, так что вашим пользователям не нужно видеть простые и скучные значения по умолчанию. В качестве примера давайте определим пользовательские страницы ошибок для ошибок HTTP 404 и 500, двух наиболее распространенных. Определение страниц для других ошибок работает одинаково.

Чтобы объявить пользовательский обработчик ошибок, используется декоратор @errorhandler. Я собираюсь поместить обработчики ошибок в новый модуль app/errors.py.

from flask import render_template
from app import app, db

@app.errorhandler(404)
def not_found_error(error):
    return render_template('404.html'), 404

@app.errorhandler(500)
def internal_error(error):
    db.session.rollback()
    return render_template('500.html'), 500

Функции ошибок работают аналогично функциям просмотра. Для этих двух ошибок я возвращаю содержимое их соответствующих шаблонов. Обратите внимание, что обе функции возвращают второе значение после шаблона, который является номером кода ошибки. Для всех функций представления, которые я создал до сих пор, мне не нужно было добавлять второе возвращаемое значение, потому что по умолчанию 200 (код состояния для успешного завершения) — это то, что я хотел. Сейчас это страницы с ошибками, поэтому я хочу, чтобы код состояния ответа это отражал.

Обработчик ошибок для 500-й ошибки может быть вызван после возникновения сбоя базы данных, которая на самом деле была вызвана умышленным случаем дубликата имени пользователя. Чтобы убедиться, что неудачные сеансы базы данных не мешают доступу к базе данных, вызванным шаблоном, я выдаю откат сеанса. Это сбрасывает сеанс в чистое состояние.

Вот шаблон для ошибки 404:

{% extends "base.html" %}

{% block content %}
    <h1>File Not Found</h1>
    <p><a href="{{ url_for('index') }}">Back</a></p>
{% endblock %}

И вот одна из ошибок 500:

{% extends "base.html" %}

{% block content %}
    <h1>An unexpected error has occurred</h1>
    <p>The administrator has been notified. Sorry for the inconvenience!</p>
    <p><a href="{{ url_for('index') }}">Back</a></p>
{% endblock %}

Оба шаблона наследуют шаблон base.html, так что страница с ошибками имеет тот же внешний вид, что и обычные страницы приложения.

Чтобы получить эти обработчики ошибок, зарегистрированные в Flask, мне нужно импортировать новый модуль app/errors.py после создания экземпляра приложения:

# ...

from app import routes, models, errors

Если вы установили FLASK_DEBUG = 0 в сеансе терминала и затем снова вызвали ошибку повторного имени пользователя, вы увидите более приятную страницу с ошибкой.

Или так! Рекомендую придумать что то свое в качестве упражнения.

Отправка ошибок по электронной почте

Другая проблема с обработкой ошибок по умолчанию, предоставляемой Flask, заключается в том, что нет уведомлений! Трассировка стека ошибки печатается на терминале, а это означает, что вывод процесса сервера должен контролироваться на обнаружение ошибок. Когда вы запускаете приложение во время разработки, это нормально, но как только приложение будет развернуто на production сервере, никто не будет смотреть на результат, поэтому необходимо создать более надежное решение.

Я думаю, что очень важно, чтобы я активно реагировал на ошибки. Если в production версии приложения возникает ошибка, я хочу знать сразу. Таким образом, моим первым решением будет сконфигурировать Flask для отправки мне сообщения по email сразу после возникновения ошибки с трассировкой стека ошибки в сообщении электронной почты.

Первым шагом является добавление данных сервера электронной почты в файл конфигурации:

class Config(object):
    # ...
    MAIL_SERVER = os.environ.get('MAIL_SERVER')
    MAIL_PORT = int(os.environ.get('MAIL_PORT') or 25)
    MAIL_USE_TLS = os.environ.get('MAIL_USE_TLS') is not None
    MAIL_USERNAME = os.environ.get('MAIL_USERNAME')
    MAIL_PASSWORD = os.environ.get('MAIL_PASSWORD')
    ADMINS = ['your-email@example.com']

Переменные конфигурации для электронной почты содержат сервер и порт, флаг для включения зашифрованных соединений и необязательное имя пользователя и пароль. Пять переменных конфигурации получены из их сопоставлений переменным среды. Если сервер электронной почты не установлен в среде, то я буду использовать это как знак того, что ошибки электронной почты должны быть отключены. Порт сервера электронной почты также можно указать в переменной среды, но если он не установлен, используется стандартный порт 25. Учетные данные почтового сервера по умолчанию не используются, но могут быть предоставлены при необходимости. Переменная конфигурации ADMINS представляет собой список адресов электронной почты, которые будут получать отчеты об ошибках, поэтому ваш собственный адрес электронной почты должен быть в этом списке.

Flask использует пакет logging Python для ведения своих журналов, а этот пакет уже имеет возможность отправлять журналы по электронной почте. Все, что мне нужно сделать, чтобы отправлять электронные сообщения, содержащие ошибки, — это добавить экземпляр SMTPHandler в объект журнала Flask, которым является app.logger:

import logging
from logging.handlers import SMTPHandler

# ...

if not app.debug:
    if app.config['MAIL_SERVER']:
        auth = None
        if app.config['MAIL_USERNAME'] or app.config['MAIL_PASSWORD']:
            auth = (app.config['MAIL_USERNAME'], app.config['MAIL_PASSWORD'])
        secure = None
        if app.config['MAIL_USE_TLS']:
            secure = ()
        mail_handler = SMTPHandler(
            mailhost=(app.config['MAIL_SERVER'], app.config['MAIL_PORT']),
            fromaddr='no-reply@' + app.config['MAIL_SERVER'],
            toaddrs=app.config['ADMINS'], subject='Microblog Failure',
            credentials=auth, secure=secure)
        mail_handler.setLevel(logging.ERROR)
        app.logger.addHandler(mail_handler)

Как видно, я включил регистратор электронной почты только, когда приложение работает без режима отладки, что определено приложением в app.debug как True, а также когда сервер электронной почты существует в конфигурации.

Настройка почтового регистратора несколько утомительна из-за необходимости обрабатывать дополнительные параметры безопасности, которые присутствуют на многих серверах электронной почты. Но в сущности, вышеприведенный код создает экземпляр SMTPHandler, устанавливает его уровень, чтобы он отправлял только сообщения об ошибках, а не предупреждения, информационные или отладочные сообщения и, наконец, прикреплял их к app.logger из Flask.

Существует два подхода к проверке работоспособности этой функции. Самый простой способ — использовать SMTP-сервер отладки от Python. Это ложный почтовый сервер, который принимает сообщения электронной почты, но вместо их отправки выводит их на консоль. Чтобы запустить этот сервер, откройте второй сеанс терминала и запустите на нем следующую команду:

(venv) $ python -m smtpd -n -c DebuggingServer localhost:8025

Оставьте запущенный SMTP-сервер отладки и вернитесь к своему первому терминалу и установите export MAIL_SERVER = localhost и MAIL_PORT = 8025 (используйте set вместо export, если вы используете Microsoft Windows). Убедитесь, что для переменной FLASK_DEBUG установлено значение 0 или не установлено вообще, так как приложение не будет отправлять электронные письма в режиме отладки.
Запустите приложение и вызовите ошибку SQLAlchemy еще раз, чтобы узнать, как сеанс терминала, на котором работает поддельный почтовый сервер, показывает электронное письмо с полным содержимым стека ошибки.

Второй метод тестирования для этой функции — настроить настоящий почтовый сервер. Ниже приведена конфигурация для использования почтового сервера для учетной записи Gmail:

export MAIL_SERVER=smtp.googlemail.com
export MAIL_PORT=587
export MAIL_USE_TLS=1
export MAIL_USERNAME=<your-gmail-username>
export MAIL_PASSWORD=<your-gmail-password>

Если вы используете Microsoft Windows, не забудьте использовать set вместо export в каждой из приведенной выше инструкции.

Функции безопасности вашей учетной записи Gmail могут препятствовать приложению отправлять электронную почту через нее, если вы явно не разрешаете «less secure apps» («менее безопасным приложениям») доступ к вашей учетной записи Gmail. Прочитать об этом можно здесь, и если вас беспокоит безопасность вашей учетной записи, можно создать вторичную учетную запись, которую настройте только для проверки электронной почты, или временно включите разрешение для менее безопасных приложений на время запуска этого теста, а затем вернитесь к умолчанию.

Запись лога в файл

Получение ошибок по электронной почте полезно, но иногда недостаточно. Есть некоторые случаи сбоя, которые не описываются исключением Python и не являются серьезной проблемой, но они все равно могут быть достаточно интересными для сохранения в целях отладки. По этой причине я также буду поддерживать логфайл для приложения.

Чтобы включить ведение журнала другого обработчика, на этот раз типа RotatingFileHandler необходимо включить logger приложения аналогично обработчику электронной почты.

# ...
from logging.handlers import RotatingFileHandler
import os

# ...

if not app.debug:
    # ...

    if not os.path.exists('logs'):
        os.mkdir('logs')
    file_handler = RotatingFileHandler('logs/microblog.log', maxBytes=10240,
                                       backupCount=10)
    file_handler.setFormatter(logging.Formatter(
        '%(asctime)s %(levelname)s: %(message)s [in %(pathname)s:%(lineno)d]'))
    file_handler.setLevel(logging.INFO)
    app.logger.addHandler(file_handler)

    app.logger.setLevel(logging.INFO)
    app.logger.info('Microblog startup')

Я пишу логфайл с именем microblog.log в каталоге logs, который я создаю, если он еще не существует.

Класс RotatingFileHandler удобен, потому что он переписывает журналы, гарантируя, что файлы журнала не будут слишком большими, если приложение работает в течение длительного времени. В этом случае я ограничиваю размер логфайла 10 КБ, и храню последние десять файлов журнала в качестве резервных копий.

Класс logging.Formatter предоставляет настройку формата сообщений журнала. Поскольку эти сообщения отправляются в файл, я хочу, чтобы они содержали как можно больше информации. Поэтому я использую формат, который включает отметку времени, уровень ведения журнала,
сообщение, исходный файл и номер строки, откуда возникла запись в журнале.

Чтобы сделать регистрацию более полезной, я также понижаю уровень ведения журнала до категории INFO, как в регистраторе приложений, так и в обработчике файлов. Если вы не знакомы с категориями ведения журнала, это DEBUG, INFO, WARNING,ERROR и CRITICAL в порядке возрастания степени тяжести.

В качестве первого полезного использования логфайла сервер записывает строку в журнал каждый раз, когда он запускается. Когда приложение запускается на production сервере, эти записи журнала сообщают вам, когда сервер был перезапущен.

Исправление дубля имени пользователя

Я слишком долго использовал ошибку дублирования имени пользователя. Теперь, когда я показал вам, как подготовить приложение для обработки подобных ошибок, я могу наконец-то это исправить.

Если вы помните, RegistrationForm уже выполняет проверку для имен пользователей, но требования формы редактирования немного отличаются. Во время регистрации мне нужно убедиться, что имя пользователя, введенное в форму, не существует в базе данных. В форме профиля редактирования я должен выполнить ту же проверку, но с одним исключением. Если пользователь оставляет исходное имя пользователя нетронутым, то проверка должна его разрешить, поскольку это имя пользователя уже назначено этому пользователю. Ниже вы можете увидеть, как я выполнил проверку имени пользователя для этой формы:

class EditProfileForm(FlaskForm):
    username = StringField('Username', validators=[DataRequired()])
    about_me = TextAreaField('About me', validators=[Length(min=0, max=140)])
    submit = SubmitField('Submit')

    def __init__(self, original_username, *args, **kwargs):
        super(EditProfileForm, self).__init__(*args, **kwargs)
        self.original_username = original_username

    def validate_username(self, username):
        if username.data != self.original_username:
            user = User.query.filter_by(username=self.username.data).first()
            if user is not None:
                raise ValidationError('Please use a different username.')

Реализация выполняется в специальном методе проверки, функция super в конструкторе класса, который принимает исходное имя пользователя в качестве аргумента. Это имя пользователя сохраняется как переменная экземпляра и проверяется в методе validate_username(). Если имя пользователя, введенное в форму, совпадает с исходным именем пользователя, то нет причин проверять базу данных на наличие дубликатов.

Чтобы использовать этот новый метод проверки, мне нужно добавить исходный аргумент имени пользователя в функцию вида, где создается объект формы:

@app.route('/edit_profile', methods=['GET', 'POST'])
@login_required
def edit_profile():
    form = EditProfileForm(current_user.username)
    # ...

Теперь ошибка исправлена, и дубликаты в форме профиля редактирования будут предотвращены в большинстве случаев. Это не идеальное решение, поскольку оно может не работать, когда два или несколько процессов одновременно обращаются к базе данных. В этой ситуации состояние гонки может привести к валидации, но спустя мгновение при попытке переименования база данных уже была изменена другим процессом и не может переименовать пользователя. Это несколько маловероятно, за исключением очень занятых приложений, у которых много серверных процессов, поэтому я пока не буду беспокоиться об этом.

На этом этапе вы можете попытаться воспроизвести ошибку еще раз, чтобы увидеть, как ее предотвращает метод проверки формы.

<<< предыдущая следующая >>>

P.S.

Работа над ошибками

От переводчика

Решил я проверить получение сообщений ошибки админу на почту. Для этого я испортил модуль routes.py. Для этой самой «порчи», я закомментировал декоратор @app.route('/edit_profile', methods=['GET', 'POST']) перед def edit_profile(). В итоге получил ошибку и в файл лога все это вывалилось, а вот письмо не прилетело. Я использую Python 3.3. Возможно в более новых версиях этого и не случится. Но в Windows 7 с русской раскладкой это случилось.

При попытке отправить сообщение админу приложение получило ошибку кодировки при формировании сообщения. В окне консоли содержались такие строки:

Как видим ссылка указывает на директорию в стандартном питоне, а не в виртуальном окружении.

logging в 3-й версии является стандартной библиотекой Python, поэтому вам не нужно устанавливать ее используя pip.

Про стандартные модули

И модуль протоколирования, который вы можете найти в PyPI, устаревший, а не Python3-совместимый.

(Согласно файлу README его последняя версия была выпущена 02 марта 2005 года.)

Поэтому просто не пытайтесь установить logging.
Возьмите новый модуль в стандартной библиотеке как должное. Если вам принципиально использовать его в виртальной библиотеке.

После копии в venvLib logging импортируется из виртуальной среды

Еще раз получаю ошибку

logging теперь виртуальный. А вот smtplib стандартный.

Не думаю, что надо тащить все библиотеки из стандартной среды в виртуальную.
Ошибка от этого не исчезнет.

Про стандартный модуль email

Проблема с кодировкой в сообщении решается использованием стандартного пакета email для создания сообщения с указанием предпочитаемой кодировки.

Вот пример с просторов интернета для этого пакета :

# -*- coding: utf-8 -*-
from email.mime.multipart import MIMEMultipart
from email.mime.text import MIMEText
import quopri
def QuoHead(String):
    s = quopri.encodestring(String.encode('UTF-8'), 1, 0)
    return "=?utf-8?Q?" + s.decode('UTF-8') + "?="
FIOin = "Хрюша Степашкин"
emailout = "some@test.ru"
emailin = "some2@test.ru"
msg = MIMEMultipart()
msg["Subject"] = QuoHead("Добрый день " + FIOin).replace('=n', '')
msg["From"] = (QuoHead("Каркуша Федоровна") + "  <" + emailout + ">").replace('=n', '') 
msg["To"] = (QuoHead(FIOin) + "  <" + emailin + ">").replace('=n', '')
m = """Добрый день.
  Это тестовое письмо.
Пожалуйста, не отвечайте на него."""
text = MIMEText(m.encode('utf-8'), 'plain', 'UTF-8')
msg.attach(text)
print(msg.as_string())

Но, как это применить для отправки сообщений об ошибке?!
Может кто-то предложит в комментариях к статье.

В модуле flask-mail эта ситуевина вроде как поправлена. Но тут используется logging и smtplib

В итоге пока так. Поправил я строку в модуле smtplib.py .

Добавил encode('utf-8')

И после перезапуска сервера при искусственной ошибке я, наконец-то, получил сообщение на почту.

<<< предыдущая следующая >>>

Источник

Как вы обрабатываете ошибки в SQLAlchemy? Я относительно новичок в SQLAlchemy и пока не знаю.

Прежде чем использовать SQLAlchemy, я бы сделал что-то вроде

status = db.query("INSERT INTO users ...")
if (!status):
    raise Error, db.error

Но теперь я кодирую в SQLAlchemy, и я делаю что-то вроде

user = User('Boda Cydo')
session.add(user)
session.commit()

Нет ошибок при проверке!

Мне не нравится этот стиль кодирования без проверки ошибок.

Проконсультируйтесь о том, как проверять и обрабатывать ошибки в SQLAlchemy!

С уважением, Бода Сидо.

Ответ 1

Ваш пример говорит:

status = db.query("INSERT INTO users ...")
if (!status):
    raise Error, db.error

Похоже, это означает, что вы хотите вызвать исключение, если в запросе есть какая-то ошибка (с raise Error, db.error). Однако sqlalchemy уже делает это для вас — так

user = User('Boda Cydo')
session.add(user)
session.commit()

Это то же самое. Часть проверки и повышения уже находится внутри SQLAlchemy.

Вот список ошибок, которые может вызвать sqlalchemy, взятых из help(sqlalchemy.exc) и help(sqlalchemy.orm.exc):

sqlalchemy.exc:
- ArgumentError — Возникает, когда указан неверный или конфликтующий аргумент функции.
  Эта ошибка обычно соответствует ошибкам состояния времени строительства.
- CircularDependencyError — Возникает при топологических сортировках при обнаружении циклической зависимости
- CompileError — возникает при возникновении ошибки во время компиляции SQL
- ConcurrentModificationError
- DBAPIError — Возникает при сбое выполнения операции базы данных.
  Если во время выполнения SQL произошла операция поднятия ошибки
  Заявление, что выражение и его параметры будут доступны на
  объект исключения в атрибутах statement и params.
  Обернутый объект исключения доступен в атрибуте orig.
  Его тип и свойства зависят от реализации DB-API.
- DataError Обертывает DB-API DataError.
- DatabaseError — переносит DB-API DatabaseError.
- DisconnectionError — Обнаружено разъединение в необработанном соединении DB-API.
  быть поднятым PoolListener, чтобы пул хостов принудительно отключался.
- FlushError
- IdentifierError — Возникает, когда имя схемы превышает максимально допустимое количество символов
- IntegrityError — переносит DB-API IntegrityError.
- InterfaceError — Обертка DB-API InterfaceError.
- InternalError — переносит DB-API InternalError.
- InvalidRequestError — SQLAlchemy попросили сделать что-то, чего он не может. Эта ошибка обычно соответствует ошибкам состояния во время выполнения.
- NoReferenceError — Поднято ForeignKey, чтобы указать, что ссылка не может быть разрешена.
- NoReferencedColumnError — Вызывается ForeignKey, когда упомянутый Column не может быть найден.
- NoReferencedTableError — Вызывается ForeignKey, когда упомянутый Table не может быть найден.
- NoSuchColumnError — из RowProxy запрашивается несуществующий столбец.
- NoSuchTableError — Таблица не существует или не видна для соединения.
- NotSupportedError — переносит DB-API NotSupportedError.
- OperationalError — переносит DB-API OperationalError.
- ProgrammingError — переносит DB-API ProgrammingError.
- SADeprecationWarning — выдается один раз за использование устаревшего API.
- SAPendingDeprecationWarning — выдается один раз за использование устаревшего API.
- SAWarning — Выдается во время выполнения.
- SQLAlchemyError — Общий класс ошибок.
- SQLError — Возникает при сбое выполнения операции с базой данных.
- TimeoutError — Возникает, когда пул соединений теряет время при получении соединения.
- UnboundExecutionError — была предпринята попытка выполнить SQL без подключения к базе данных.
- UnmappedColumnError
sqlalchemy.orm.exc:
- ConcurrentModificationError — Ряды были изменены за пределами единицы работы.
- FlushError — во время flush() обнаружено недопустимое состояние.
- MultipleResultsFound — Требуется один результат для базы данных, но найдено более одного.
- NoResultFound — Требуется результат для базы данных, но ничего не найдено.
- ObjectDeletedError — Операция refresh() не смогла повторно получить строку объекта.
- UnmappedClassError — Запрошена операция сопоставления для неизвестного класса.
- UnmappedColumnError — Операция сопоставления была запрошена для неизвестного столбца.
- UnmappedError — TODO
- UnmappedInstanceError — Операция сопоставления была запрошена для неизвестного экземпляра.

Ответ 2

Я попробовал это, и он показал мне конкретное сообщение об ошибке.

from sqlalchemy.exc import SQLAlchemyError

try:
# try something

except SQLAlchemyError as e:
  error = str(e.__dict__['orig'])
  return error

Надеюсь это поможет

Ответ 3

SQLAlchemy вызовет исключение при ошибке….

Источник