Python sqlite3 模块，dbapi2() 实例源码

def create_connect_args(self, url):
        if url.username or url.password or url.host or url.port:
            raise exc.ArgumentError(
                "Invalid SQLite URL: %s\n"
                "Valid SQLite URL forms are:\n"
                " sqlite:///:memory: (or, sqlite://)\n"
                " sqlite:///relative/path/to/file.db\n"
                " sqlite:////absolute/path/to/file.db" % (url,))
        filename = url.database or ':memory:'
        if filename != ':memory:':
            filename = os.path.abspath(filename)

        opts = url.query.copy()
        util.coerce_kw_type(opts, 'timeout', float)
        util.coerce_kw_type(opts, 'isolation_level', str)
        util.coerce_kw_type(opts, 'detect_types', int)
        util.coerce_kw_type(opts, 'check_same_thread', bool)
        util.coerce_kw_type(opts, 'cached_statements', int)

        return ([filename], opts)

def patch():
    wrapped = wrapt.FunctionWrapper(_connect, traced_connect)

    setattr(sqlite3, 'connect', wrapped)
    setattr(sqlite3.dbapi2, 'connect', wrapped)

def unpatch():
    sqlite3.connect = _connect
    sqlite3.dbapi2.connect = _connect

def dbapi(cls):
        try:
            from pysqlite2 import dbapi2 as sqlite
        except ImportError, e:
            try:
                from sqlite3 import dbapi2 as sqlite  # try 2.5+ stdlib name.
            except ImportError:
                raise e
        return sqlite

def __init__(self, database_dir='./output'):
        if not os.path.exists(database_dir):
            os.makedirs(database_dir)

        path = os.path.join(database_dir, DATABASE_NAME)

        self.engine = sqlalchemy.create_engine('sqlite:///'+path, module=sqlite, echo=False)
        self.Sessionmaker = sessionmaker(bind=self.engine)

        Base = declarative_base()
        class PostORM(Base):
            __tablename__ = 'posts'
            __table_args__ = {'sqlite_autoincrement': True}

            id = sqlalchemy.Column(sqlalchemy.Integer, primary_key=True)
            name = sqlalchemy.Column(sqlalchemy.String)
            password = sqlalchemy.Column(sqlalchemy.String)
            title = sqlalchemy.Column(sqlalchemy.String)
            author = sqlalchemy.Column(sqlalchemy.String)
            url = sqlalchemy.Column(sqlalchemy.String)
            shortlink = sqlalchemy.Column(sqlalchemy.String)
            subreddit = sqlalchemy.Column(sqlalchemy.String)
            folder = sqlalchemy.Column(sqlalchemy.String)
            num_files = sqlalchemy.Column(sqlalchemy.String)
            date_added = sqlalchemy.Column(DATETIME)

            def __init__(self, title, author, url, shortlink, subreddit, num_files, folder):
                self.title = title
                self.author = author
                self.url = url
                self.shortlink = shortlink
                self.subreddit = subreddit
                self.folder = folder
                self.num_files = num_files
                self.date_added = datetime.now()

        self.PostORM = PostORM

        self.Base = Base
        if not self.engine.dialect.has_table(self.engine, PostORM.__tablename__):
            self.Base.metadata.create_all(self.engine)
        self.session = self.Sessionmaker()