Linux+python+odbc+SQLServer访问方式

安装ODBC库wget http://www.unixodbc.org/unixODBC-2.3.7.tar.gztar zxvf unixODBC-2.3.7.tar.gzcd unixODBC-2.3.7./configuremakemake install添加配置vi odbc.ini[SQLTest]Driver = FreeTDSDe...

韩搏

45836人浏览 · 2020-02-05 16:26:25

韩搏 · 2020-02-05 16:26:25 发布

安装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

安装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

连接测试

	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!

安装py依赖包

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

测试代码

#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()