連接postgresql

# psycopg2

engine=create_engine('postgresql+psycopg2://scott:tiger@localhost/mydatabase')#

python 連接postgresql使用psycopg2作為默認的DBAPI

The first time a method like?Engine.execute()orEngine.connect()is called, the?Engine?establishes a real?DBAPI?connection to the database, which is then used to emit the SQL.

Thecreate_engine()function produces anEngineobject basedon a URL.

?1?from?sqlalchemy.engine?import?create_engine
?2?from?sqlalchemy.schema?import?MetaData,?Table,?Column,?ForeignKey,?Sequence
?3?from?sqlalchemy.types?import?*
?4?
?5?engine?=?create_engine('postgres://test:test@localhost/test',?echo=True)
?6?
?7?metadata?=?MetaData()
?8?metadata.bind?=?engine
?9?
10?book_table?=?Table('book',?metadata,
11?????Column('id',?Integer,?Sequence('seq_pk'),?primary_key=True),
12?????Column('title',?Unicode(255),?nullable=False),
13?)
14?
15?author_table?=?Table('author',?metadata,
16?????Column('id',?Integer,?Sequence('seq_pk'),?primary_key=True),
17?????Column('name',?Unicode(255),?nullable=False),
18?)
19?
20?bookauthor_table?=?Table('bookauthor',?metadata,
21??? Column('book_id',?Integer,?ForeignKey('book.id'),?nullable=False),
22??? Column('author_id',?Integer,?ForeignKey('author.id'),?nullable=False),
23)
24
25metadata.create_all(checkfirst=True)

首先我們還是create_engine，然后新建一個MetaData對象，把engine綁上去，接下來，開始在metadata中定義表結構(metadata由Table構造函數傳入)，我們這里定義了3張表，分別是book、author和bookauthor關系表(“多對多”)，其中新建一個Sequence對象，專門處理主鍵生成。最后我們通過執行metadata.create_all()創建數據庫表，參數checkfirst=True表示如果數據庫相關對象已經存在，則不重復執行創建。

對于已經存在于數據庫中的表，我們可以通過傳入autoload=True參數到Table構造函數的方式來加載現有的表結構到metadata中，而不必挨個兒再寫一遍Column清單。

看到這兒，你也許覺得挺麻煩，不是么？Django和RoR都是可以直接定義數據model類，順帶就把schema也定義了，而不是像這樣單獨去寫表結構的schema，顯得很"底層"。確實，這樣用SQLAlchemy并不是最優化的，SQLAlchemy本身并不會自動的幫你做很多事，但它基礎打得很牢。如果你感興趣，也可以先去看一下SQLAlchemy的擴展模塊Elixir，通過Elixir，你可以像Ruby on Rails那樣定義出實體和關系("Active Record")。

1. 文／人世間（簡書作者）
2. 原文鏈接：http://www.jianshu.com/p/e6bba189fcbd
3. 著作權歸作者所有，轉載請聯系作者獲得授權，并標注“簡書作者”。
5. # -*- coding: utf-8 -*-
6. __author__ = 'ghost'
8. from sqlalchemy import create_engine, Table, Column, Integer, String, MetaData, ForeignKey
9. # 連接數據庫
10. engine = create_engine("mysql://root:@localhost:3306/webpy?charset=utf8",encoding="utf-8", echo=True)
11. # 獲取元數據
12. metadata = MetaData()
13. # 定義表
14. user = Table('user', metadata,
15. Column('id', Integer, primary_key=True),
16. Column('name', String(20)),
17. Column('fullname', String(40)),
18. )
20. address = Table('address', metadata,
21. Column('id', Integer, primary_key=True),
22. Column('user_id', None, ForeignKey('user.id')),
23. Column('email', String(60), nullable=False)
24. )
25. # 創建數據表，如果數據表存在，則忽視
26. metadata.create_all(engine)
27. # 獲取數據庫連接
28. conn = engine.connect()

sqlalchemy連接postgresql數據庫

import sqlalchemy
import pnosql

class Confsql:
    def __init__(self,dbstr="postgresql+psycopg2://postgres:root@localhost:5432/Usermodel"):

        self.engine = sqlalchemy.create_engine(dbstr, echo=True)
        self.metadata = sqlalchemy.MetaData()
        self.metadata.bind = self.engine

def runquery(self, sqlstr):
        s = sqlstr
        result = self.engine.execute(sqlstr)
        rows = result.fetchall()
        result.close()

需要對返回的數據進行修改才行

def runsp(self,sqlstr):
    s = sqlstr
    result = self.engine.execute(sqlstr)
    rows = result.fetchall()
    result.close()

    result  = []
    for row in rows:
        x = {}
        x["barcode"] = row[0]
        x["spcode"] = row[1]
        x["spname"] = row[2]
        x["spformat"] = row[3]
        x["height"] = row[4]
        x["width"] = row[5]
        x["thickness"] = row[6]

        x["comp"] = 'youke'
        x["parentcomp"] = 'yz'
        x["_id"] = str(uuid.uuid1())

        result.append(x)

    return result

SqlAlchemy應用

1. from sqlalchemy import create_engine,MetaData,Table,select
3. engine = create_engine('postgresql+psycopg2://postgres:root@localhost:5432/blogdb')
4. metadata = MetaData()
5. metadata.bind = engine
6. auth_permission = Table('auth_permission',metadata,autoload = True)

查詢操作

1. def query_code(codename):
2. info = {'name':'','codename':''}
3. s = select([auth_permission.c.codename, auth_permission.c.name, ]).where(auth_permission.c.codename == codename)
4. codename_query = engine.execute(s)
5. for row in codename_query:
6. info['codename'] = row[0]
7. info['name'] = row[1]
8. codename_query.close()
9. return info

修改操作

1. #修改權限
2. def updata(codename,name):
3. s = auth_permission.update().where(auth_permission.c.codename == codename).values(name=name,codename=codename)
4. c = engine.execute(s)
5. c.close()

添加操作

1. # 添加權限
2. def add(codename,name,content_type_id):
3. s = auth_permission.insert().values(name=name,codename=codename,content_type_id=content_type_id)
4. c = engine.execute(s)
5. c.close()

刪除操作

1. # 刪除權限
2. def delete(codename):
3. s = auth_permission.delete().where(auth_permission.c.codename == codename)
4. c = engine.execute(s)
5. c.close()

來自為知筆記(Wiz)