definit_app(self,app):"""This callback can be used to initialize an application for the use with this database setup. Never use a database in the context of an application not initialized that way or connections will leak. """# 首先是尝试获取app中的配置,如果没有找到则发出警告if ('SQLALCHEMY_DATABASE_URI'notin app.config and# 如果有多个数据库,需要配置这个选项'SQLALCHEMY_BINDS'notin app.config ): warnings.warn('Neither SQLALCHEMY_DATABASE_URI nor SQLALCHEMY_BINDS is set. ''Defaulting SQLALCHEMY_DATABASE_URI to "sqlite:///:memory:".' )# 防御性编程,给dict设置一些默认值# setdefault是dict的默认值 app.config.setdefault('SQLALCHEMY_DATABASE_URI', 'sqlite:///:memory:') app.config.setdefault('SQLALCHEMY_BINDS', None) app.config.setdefault('SQLALCHEMY_NATIVE_UNICODE', None) app.config.setdefault('SQLALCHEMY_ECHO', False) app.config.setdefault('SQLALCHEMY_RECORD_QUERIES', None) app.config.setdefault('SQLALCHEMY_POOL_SIZE', None) app.config.setdefault('SQLALCHEMY_POOL_TIMEOUT', None) app.config.setdefault('SQLALCHEMY_POOL_RECYCLE', None) app.config.setdefault('SQLALCHEMY_MAX_OVERFLOW', None) app.config.setdefault('SQLALCHEMY_COMMIT_ON_TEARDOWN', False) track_modifications = app.config.setdefault('SQLALCHEMY_TRACK_MODIFICATIONS', None )if track_modifications isNone: warnings.warn(FSADeprecationWarning('SQLALCHEMY_TRACK_MODIFICATIONS adds significant overhead and ''will be disabled by default in the future. Set it to True ''or False to suppress this warning.' )) app.extensions['sqlalchemy']=_SQLAlchemyState(self)@app.teardown_appcontextdefshutdown_session(response_or_exc):if app.config['SQLALCHEMY_COMMIT_ON_TEARDOWN']:if response_or_exc isNone: self.session.commit() self.session.remove()return response_or_exc
create_app 方法的源码
def_execute_for_all_tables(self,app,bind,operation,skip_tables=False): app = self.get_app(app)if bind =='__all__': binds = [None] +list(app.config.get('SQLALCHEMY_BINDS') or ())elifisinstance(bind, string_types)or bind isNone: binds = [bind]else: binds = bindfor bind in binds: extra ={}ifnot skip_tables: tables = self.get_tables_for_bind(bind) extra['tables']= tables op =getattr(self.Model.metadata, operation)op(bind=self.get_engine(app, bind), **extra)defcreate_all(self,bind='__all__',app=None):"""Creates all tables. .. versionchanged:: 0.12 Parameters were added """ self._execute_for_all_tables(app, bind, 'create_all')
defget_app(self,reference_app=None):"""Helper method that implements the logic to look up an application."""# 如果关键字参数的app不为空,就返回参数的appif reference_app isnotNone:return reference_app# 如果current_app不为空,则返回current_appif current_app:return current_app._get_current_object()# 如果对象的app属性不为空,则返回对象的app属性if self.app isnotNone:return self.appraiseRuntimeError('No application found. Either work inside a view function or push'' an application context. See'' http://flask-sqlalchemy.pocoo.org/contexts/.' )