All,
Update: based on google result and answer, I added more hints, still not finished.
In using sqlite3 and during study of sqlalchemy, I found it is necessary to write below code for those housekeeping purpose for managing data, however, it may be a hard part for me to doing that in sqlalchemy then I turning back to sqlite3 module.
Below code lists 10 more steps as housekeeping jobs and most of them came from WEB, I doubt someone with expertise can checking and padding the missing part for it. And if someone know how to do it in SQLAlchemy, could you also sharing it pls?
1. testing if the database file existing
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. testing if database file is a valid sqlite3 format
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]/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. check table exist
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.backup database file in disk
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. backup table - in same database file
c=conn.cursor()
c.execute("CREATE TABLE demo_backup AS SELECT * FROM demo;")
6. rename table
c.execute("ALTER TABLE foo RENAME TO bar;")
7. copy table to/from different database:
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]/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 test database is locked or not?
#possible?
try:
c = sqlite.connect(database_name, timeout=0)
c.commit()
except OperationalError # OperationalError: database is locked
9. timeout to connect to database, waiting other invoker release the lock
c = sqlite.connect(database_name, timeout=30.0) # default 5sec
10 force all database connections release/commit A.K.A to release all lock?
refer #12
11. multi-threads in using sqlite in python:
http://code.activestate.com/recipes/526618/
http://www.yeraze.com/2009/01/python-sqlite-multiple-threads/
12 get conn from SQLAlchemy?
#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**
Thanks for your time!
Rgs, KC