Linux+python+odbc+SQLServer访问方式

蔚落 2023-07-03 10:41 8阅读 0赞

1.  安装ODBC库

wget http://www.unixodbc.org/unixODBC-2.3.7.tar.gz
    	tar zxvf unixODBC-2.3.7.tar.gz
    	cd unixODBC-2.3.7 
    	./configure
    	make
    	make install
    
    	添加配置
    	vi /usr/local/etc/odbc.ini 
    	[SQLTest]
    	Driver = FreeTDS
    	Description = OBDC FreeTDS
    	Database = db_test
    	Servername = SQLTest
    	
    	vi /usr/local/etc/odbcinst.ini 
    	[FreeTDS]
    	Description = ODBC for FreeTDS
    	Driver = /usr/local/lib/libtdsodbc.so
    	Setup = /usr/lib/x86_64-linux-gnu/odbc/libtdsS.so

1.  安装freetds(驱动SQLServer)

wget ftp://ftp.freetds.org/pub/freetds/stable/freetds-0.91.100.tar.gz
    	./configure --with-sdsver=8.0 --with-unixODBC=/usr/local  # --enable-msdblib=yes  C语言使用
    	make
    	make install
    	
      
    	添加配置
    	vi /usr/local/etc/freetds.conf最后添加
    	[SQLTest]
    		host = 192.168.2.9
    		port = 1433
    		tds version = 8.0

1.  连接测试

tsql -H 192.168.2.9 -p 1433 -U sa -P sa  ==> 1>
    	tsql -S SQLTest -U sa   ===> 1>
    	odbcinst -q -s ==> [SQLTest]
    	isql -v SQLTest 用户名 密码  ==> Connected!

1.  安装py依赖包

wget https://files.pythonhosted.org/packages/a9/b1/0f91ec8a8357272dd988f20958467dded751d720ca2f0bdd37b7e117d941/pyodbc-3.0.7.zip
    	python setup.py build
    	python setup.py install

1.  测试代码

#coding=utf8
    import string
    import pyodbc
    
    class DatabaseCursor:
        def __init__(self,conn):
            self.cursor = conn.cursor()
    
        def __enter__(self):
            return self.cursor
    
        def __exit__(self,exc_type, exc_value, exc_tb):
            self.cursor.close()
            self.cursor = None
    
    def database_connect():
        server = '192.168.2.9:1433'
        user = 'sa'
        password = 'sa'
        database = 'db_test'
        conninfo = 'DRIVER={SQLTest};SERVER=%s;port=%s;DATABASE=%s;UID=%s;PWD=%s;TDS_Version=8.0;' % (server.split(':')[0], server.split(':')[1], database, user, password)
        return pyodbc.connect(conninfo)
    
    def odbcDataToDict(cursor):
        columns = [column[0] for column in cursor.description]
        results = [dict(zip(columns,row)) for row in cursor.fetchall()]
        return results
    
    def test():
        try:
            conn = database_connect()
            with DatabaseCursor(conn) as cursor:
                cursor.execute('SELECT ID as id, name, age FROM tb_user')
                cursor = odbcDataToDict(cursor)
                for row in cursor:
                    print row
            
        except Exception as e:
            print e
        finally:
            if conn is not None:
                conn.close()
                conn = None
    
    if __name__ == '__main__':
    	test()