需要检查和填充 sqlite 数据库管理和操作代码

All,

更新：根据谷歌结果和答案，我添加了更多提示，但仍未完成。

在使用sqlite3和研究sqlalchemy的过程中，我发现有必要编写下面的代码来管理数据，但是，对我来说在sqlalchemy中做到这一点可能是一个困难的部分，然后我又回到sqlite3模块。

下面的代码列出了另外 10 个步骤作为内务工作，其中大部分来自 WEB，我怀疑有专业知识的人可以检查并填补缺失的部分。 如果有人知道如何在 SQLAlchemy 中做到这一点，您也可以分享一下吗？

1. 测试数据库文件是否存在

import sqlite3 
import os 
database_name = "newdb.db" 
if not os.path.isfile(database_name): 
    print "the database already exist" 

# connect to to db, refer #2
db_connection = sqlite3.connect(database_name) 
db_cursor = db_connection.cursor() 

2.测试数据库文件是否是有效的sqlite3格式

    http://stackoverflow.com/questions/1516508/sqlite3-in-python


    >>> c.execute("SELECT * FROM tbl") 
    Traceback (most recent call last): 
      File "<stdin>", line 1, in <module> 
    sqlite3.DatabaseError: file is encrypted or is not a database
    =========sqlalchemy way ===============
    http://www.mail-archive.com/[email protected] /cdn-cgi/l/email-protection/msg20860.html
    import os, os.path as osp
try:
    from pysqlite2 import dbapi2 as sqlite
except:
    import sqlite3 as sqlite

def isSQLite(filename):
    """True if filename is a SQLite database
    File is database if: (1) file exists, (2) length is non-zero,
                        (3) can connect, (4) has sqlite_master table
    """
    # validate file exists
    if not osp.isfile(filename):
        return False
    # is not an empty file
    if not os.stat(filename).st_size:
        return False
    # can open a connection
    try:
        conn = sqlite.connect(filename)
    except:
        return False
    # has sqlite_master
    try:
        result = conn.execute('pragma table_info(sqlite_master)').fetchall()
        if len(result) == 0:
            conn.close()
            return False
    except:
        conn.close()
        return False

    # looks like a good database
    conn.close()
    return True 

3.检查表是否存在

c=conn.cursor() 
if table_name in [row for row in c.execute("SELECT name FROM sqlite_master WHERE type='table' AND name='table_name';")]

4.备份磁盘中的数据库文件

http://stuvel.eu/archive/55/safely-copy-a-sqlite-database
import shutil, os, sqlite3
if not os.path.isdir ( backupdir ):
    raise Exception 
backupfile = os.path.join ( backupdir, os.path.basename(dbfile) + time.strftime(".%Y%m%d-%H%M") )
db = sqlite3.connect ( dbfile )
cur = db.cursor ()
cur.execute ( 'begin immediate' )
shutil.copyfile ( dbfile, backupfile )
cur.execute ( 'rollback' )
=========or========
http://github.com/husio/python-sqlite3-backup
=========or========
http://docs.python.org/release/2.6/library/sqlite3.html#sqlite3.Connection.iterdump

5. 备份表 - 在同一个数据库文件中

   c=conn.cursor() 
   c.execute("CREATE TABLE demo_backup AS SELECT * FROM demo;") 

6. 重命名表

c.execute("ALTER TABLE foo RENAME TO bar;")

7. 将表复制到不同的数据库/从不同的数据库复制表：

Thanks, MPelletier

Connect to one database 
db_connection = sqlite3.connect(database_file) 
Attach the second database
db_connection.execute("ATTACH database_file2 AS database_name2")
Insert from one to the other:
db_connection.execute("INSERT INTO FooTable SELECT * FROM database_name2.FooTable")
==========or============
db_connection.execute("INSERT INTO database_name2.FooTable SELECT * FROM FooTable")

    ========sqlalchemy way======
    http://www.mail-archive.com/[email protected] /cdn-cgi/l/email-protection/msg11563.html
      def duplicateToDisk(self, file):
    '''Tohle ulozi databazi, ktera byla pouze v pameti, na disk'''
    cur = self.connection()
    import os
    if os.path.exists(file):
      os.remove(file)
    cur.execute("attach %s as extern" % file)

    self.checkTable('extern.dictionary')
    cur.execute("insert into extern.dictionary select * from dictionary")
    cur.execute("detach extern")
    self.commit()

8 测试数据库是否被锁定？

 #possible?
 try:
    c = sqlite.connect(database_name, timeout=0)  
    c.commit() 
 except OperationalError               # OperationalError: database is locked  

9. 连接数据库超时，等待其他调用者释放锁

c = sqlite.connect(database_name, timeout=30.0)  # default 5sec

10 强制所有数据库连接释放/提交 A.K.A 释放所有锁？

   refer #12

11. python中使用sqlite的多线程：

http://code.activestate.com/recipes/526618/
http://www.yeraze.com/2009/01/python-sqlite-multiple-threads/

12 从 SQLAlchemy 获取 conn？

   #from FAQ
    #try to reuse the connection pool from SQLAlchemy
    engine = create_engine(...)
    conn = engine.connect()              #****1
    conn.connection.<do DBAPI things>
    cursor = conn.connection.cursor(<DBAPI specific arguments..>)
    ===or ==== can out of pool's manage
    conn = engine.connect()
    conn.detach()  # detaches the DBAPI connection from the connection pool
    conn.connection.<go nuts>
    conn.close()  # connection is closed for real, the pool replaces it with a new connect

    ========and not sure if this works ===========
#from sqlalchemy document                #http://www.sqlalchemy.org/docs/reference/sqlalchemy/pooling.html?highlight=connection%20pool
import sqlalchemy.pool as pool
import sqlite3 as sqlite3

conn_proxy = pool.manage(sqlite3)
# then connect normally
connection = conn_proxy.connect(...)


=====================================================================
    #****1  ： what is #****1 on above code invoked                  =_=!!
    A engine.raw_connection()
    =
    A pool.unique_connection()
    =
    A _ConnectionFairy(self).checkout()
    =
    A return _ConnectionFairy <== cls
    =   _connection_record.get_connection()
    =            _ConnectionRecord.connection
    =                return a pool.creator **which is a callable function that returns a DB-API connection object**

谢谢你的时间！

Rgs, KC

要将数据复制到不同的数据库或从不同的数据库复制数据，一般的 SQLite 方法是：

1. 连接到一个数据库

   db_connection = sqlite3.connect(database_file) 
   

2. 附加第二个数据库

   db_connection.execute("ATTACH database_file2 AS database_name2")
   

3. 从一个插入到另一个：

   db_connection.execute("INSERT INTO FooTable SELECT * FROM database_name2.FooTable")
   

   or

   db_connection.execute("INSERT INTO database_name2.FooTable SELECT * FROM FooTable")