PG8000 - Pure-Python PostgreSQL驱动程序

2021-05-14 01:19:08

PG8000配有两个API,本机PG8000 API和DB-API 2.0 StandardAPI。这些是本机API的示例,DB-API 2.0在下一部分中的示例示例。

导入pg8000,连接数据库,创建一个表,添加一些行,然后添加一个表:

>> >导入pg8000。天然>> > >> > #使用用户名Postgres&gt连接到数据库。> > >> > con = pg8000。本国的。连接(" postgres"密码=" cpsnow")>> > >> > #创建一个临时表>> > >> >骗子。跑("创建临时表书(ID序列,标题文本)")[]> > >> > #填充表格>> > >> >在(" enter' same&#34 ;," magus"):...骗局的标题运行("插入到书籍(标题)值(:title)"标题= title)[] [] []>> > >> > #打印表格中的所有行>> > >> >在骗局中排行。 run("从书中选择*,从书中选择*):...打印(行)[1," enter' s游戏"] [2,' magus&#39 ;]

>>>进口PG8000.NAID>>>>>> CON = pg8000.native.Connection(" postgres的&#34 ;,密码=" cpsnow")>>>>>> con.run("开始交易")[]>> #创建一个临时表>>> con.run("创建临时表图书(ID序列,标题文本)")[]>>>在(" enter' s游戏中的标题和#34;" Magus&#34 ;," phineas finn"):...康明("插入书(标题)值(:标题)",标题=标题)[] [] [] [] [] []>>> con.run("提交")[]>>在Con.Run中的行("从书中选择*):...打印(行)[1," Ender' s游戏"] [2,&#39 ; Magus'] [3,' phineas finn']

>>>进口PG8000.NAID>>>>>> CON = pg8000.native.Connection(" postgres的&#34 ;,密码=" cpsnow")>>>>>> #创建一个临时表>>> con.run("创建临时表图书(ID序列,标题文本)")[]>>>在(" enter' s游戏中的标题和#34;" Magus&#34 ;," phineas finn"):...康明(" INSERT INTO书(标题)VALUES(:标题)&#34 ;,标题=标题)[] [] []>>>>>> con.run("开始交易")[]>> con.run("从标题=:title",标题=" phineas finn")[]>> con.run("回滚")[]>>在Con.Run中的行("从书中选择*):...打印(行)[1," Ender' s游戏"] [2,&#39 ; Magus'] [3,' phineas finn']

>> >导入pg8000。天然>> > >> > con = pg8000。本国的。连接(" postgres"密码=" cpsnow")>> > >> >导入DateTime>> > >> > ts = datetime。日期(1980,4,27)>> >骗子。跑("选择时间戳' 2013-12-01 16:06' - :Ts",ts = ts)[[DateTime。 Timedelta(天= 12271,秒= 57960)]]

与服务器通信时,PG8000使用TheServer要求使用的字符集(客户端编码)。默认情况下,客户端编码是数据库的字符集(在创建数据库时选择),但是可以以多种方式更改,但是可以以多种方式更改(例如,postgreSQL.conf中的setcarlient_encoding)。改变ClientEncodinging的另一种方式是使用SQL命令。例如:

JSON始终从服务器释放出序列化。如果你要发送的json是一个dictthen,你可以刚才做:

>> >导入pg8000。天然>> > >> > con = pg8000。本国的。连接(" postgres"密码=" cpsnow")>> > >> > val = {':' apollo 11洞穴&#39 ;,' zebra' true,'年龄&#39 ;: 26.003}>> >骗子。运行("选择:apollo",apollo = val)[[{'年龄':26.003,'名称&#39 ;:' apollo 11洞穴' ,'斑马':true}]]

>> >进口json>> >导入pg8000。天然>> > >> > con = pg8000。本国的。连接(" postgres"密码=" cpsnow")>> > >> > >> > val = ['阿波罗11洞穴',真实,26.003]>> >骗子。 run("选择cast(:apolloasejsonb)" apollo = json。转储(val))[[' apollo 11 cave',26.003]]]

>> >导入pg8000。天然>> > >> > con = pg8000。本国的。连接(" postgres"密码=" cpsnow")>> > >> >骗子。运行("创建临时表夸克(ID序列,名称文本)")[]>> >在(''' down')的名称:...骗局。 run("插入zhark(名称)值(:name)",name = name)[] []>> > #现在执行Query>> > >> >骗子。运行("从Quark&#34选择*)[[1,' Up'],[2,#39;']]>>>>> > >> > #并重试元数据>> > >> >骗子。列[{' table_oid&#39 ;: ...,' column_attrnum&#39 ;: 1,' type_oid&#39 ;: 23,' type_size&#39 ;: 4,& #39; type_modifier&#39 ;: - 1,'格式&#39 ;: 0,'姓名&#39 ;:' id' {' table_oid' :...,' column_attrnum&#39 ;: 2,' type_oid&#39 ;: 25,' type_size&#39 ;: - 1,' type_modifier&#39 ;: - 1 ,'格式&#39 ;: 0,'姓名&#39 ;:'名称'}]>> > >> > #显示列名称>> > >> > [C ['符号名称'] con。列[' id''名称']

PostgreSQL标记存储在称为Connection.Notices并使用Append()方法添加的DEQUE。类似地,存在Connectification.notificationsand.parameter_statuses,用于更改服务器配置。的一个例子:

>> >导入pg8000。天然>> > >> > con = pg8000。本国的。连接(" postgres"密码=" cpsnow")>> > >> >骗子。跑("听aliens_landed")[]> >骗子。跑("通知Aliens_landed")[]>> > #通知是包含(后退_pid,频道,有效载荷)>&gt的元组。 > >> >骗子。通知[0](...,' aliens_landed''')

>> >导入pg8000。天然>> > >> > con = pg8000。本国的。连接(" postgres"密码=" cpsnow")>> > >> >骗子。跑(" select' silo 1' lims:lim&#34 ;,5 lim ='全部')回溯(最近呼叫最后):pg8000。例外。数据库错误: ...

相反,Docs Saytht您可以将空闲作为所有工作的替代品,这是工作:

您可能会认为以下会有效,但实际上服务器没有:

>> >导入pg8000。天然>> > >> > con = pg8000。本国的。连接(" postgres"密码=" cpsnow")>> > >> >骗子。跑("选择'孤岛1'其中' a'' v",v =(' a'和#39; b' b' 39;))回溯(最近呼叫最后):PG8000。例外。数据库错误: ...

>> >导入pg8000。天然>> > >> > con = pg8000。本国的。连接(" postgres"密码=" cpsnow")>> > >> >骗子。跑(" select' silo 1'其中' a'(选择UNNEST(:V))",v =(' a' ,' b'))[[' silo 1']]

在PostgreSQL参数中只能用于数据值,而不是标识符。有时,这可能无法按预期工作,例如以下失败:

>> >导入pg8000。天然>> > >> > con = pg8000。本国的。连接(" postgres"密码=" cpsnow")>> > >> >骗子。运行("使用密码创建用户Juan:password",password =' quail')回溯(最近调用最后):pg8000。例外。数据库错误: ...

它失败,因为PostgreSQL服务器不允许此语句才能获得参数。有许多SQL陈述,人们可能认为将有很多分析器,但是没有。

SQL Copy DateplateCan可用于复制到文件或文件等对象。在这里'一个emplyusing csv格式:

>> >导入pg8000。天然>> >来自IO导入Bytesio≫> >进口CSV>> >导入编解码器>> > >> > con = pg8000。本国的。连接(" postgres"密码=" cpsnow")>> > >> > #在内存中创建CSV文件。> > >> > Stream_in = bytesio()>> > StreamWriter =编解码器。 getwriter(' utf-8')>> > CSV_WRITER = CSV。作者(StreamWriter(Stream_in))>> > CSV_WRITER。作品([1,"电子"])>> > CSV_WRITER。作品([2," muon"])>> > CSV_WRITER。作者([3," tau"])>> > Stream_in。寻求(0)0>> > >> > #创建一个表,然后将CSV复制到IT>> > >> >骗子。跑("创建临时表Lepton(ID序列,名称文本)")[]>> >骗子。跑("从stdin复制leptin与(格式csv)" stream = stream_in)[]> > >> > #从表中复制到流>> > >> > Stream_out = bytesio()>> >骗子。运行("用(格式csv)" stream = stream_out)[]>> >流出来。寻求(0)0>> > StreamReader =编解码器。 GetReader(' UTF-8')>> >用于CSV的行。读者(StreamReader(Stream_out)):...打印(行)[' 1''电子'] [' 2'' muon&# 39;] [' 3'' tau']

如果要执行一系列SQL语句(例如,.SQL文件),则YOUCAN按预期运行它们:

>> >导入pg8000。天然>> > >> > con = pg8000。本国的。连接(" postgres"密码=" cpsnow")>> > >> >陈述="选择5;选择' eRich fromm&#39 ;;" >> > >> >骗子。 run(陈述)[[5],[' eRich fromm']]

唯一的警告是,当执行多个陈述时,您就不能有任何parameters。

假设你有一个名为我的列的列。由于它的区分大小写,因此您必须通过双引号括起来。但你做不到:

>> >导入pg8000。天然>> > >> > con = pg8000。本国的。连接(" postgres"密码=" cpsnow")>> > >> >骗子。跑("选择'你好'作为"我的列"")回溯(最近呼叫最后一个电话):syntaxError:语法无效

由于Python使用双引号来分隔字符串文字,因此一个解决方案是使用Python的三额额相级别界定字符串:

PG8000具有从Python类型到PostgreSQL类型的映射,因为它需要将SQL参数发送到服务器时。在大多数情况下,Cutp8000的默认映射旨在运行良好,但您可能希望添加Orreplace默认映射。

将Python DateTime.Timedelta对象作为PostgreSQL间隔类型发送到服务器,其中具有OID 1186.但是假设我们想创建自己的Python类作为间隔类型。然后我们已经托架了适配器:

>> >导入pg8000。天然>> > >> > con = pg8000。本国的。连接(" postgres"密码=" cpsnow")>> > >> > Class MyInterval(str):...通过和gt;> > >> > def my_interval_out(my_interval):...返回my_interval#必须返回str>> > >> >骗子。 register_out_adapter(MyInterval,1186,my_interval_out)>> >骗子。运行("选择:间隔",interval = myInterval(" 2小时"))[[DateTime。 Timedelta(秒= 7200)]]

请注意,它仍然是DateTime.Timedelta对象,因为我们唯一从Python到PostgreSQL的映射。有关HOWTO将PostgreSQL的映射更改为Python的示例,请参阅下文。

PG8000从PostgresQL类型映射到Python类型,何时收听来自服务器的QL结果。 PG8000附带的默认映射在大多数情况下,才能运行良好,但您可能希望添加或替换thedefault映射。

如果PG800收到PostgreSQL间隔类型,则具有OID 1186,它将其转换为Python DateTime.Timedelta对象。但是,让我们说我们希望创建自己的Python类来代替DateTime.Timedelta。然后我们必须注册一个适配器:

>> >导入pg8000。天然>> > >> > con = pg8000。本国的。连接(" postgres"密码=" cpsnow")>> > >> > Class MyInterval(str):...通过和gt;> > >> > def my_interval_in(my_interval_str):#参数是str的类型...返回myInterval(my_interval)>> > >> >骗子。 register_in_adapter(1186,my_interval_in)>> >骗子。跑("选择\' 2年'")[[39; 2岁']]]

请注意,注册'在'适配器仅从Python类型从博斯特格QL类型从映射上。有关如何从PostgreSQL转换为Python的示例,请参阅上面。

有时你会得到'无法确定参数的数据类型'来自服务器的errormessage:

>> >导入pg8000。天然>> > >> > con = pg8000。本国的。连接(" postgres"密码=" cpsnow")>> > >> >骗子。运行("选择:v为null",v = none)回溯(最近呼叫最后一个呼叫):pg8000。例外。 DatabaseError:{' s&#39 ;:'错误'' v&#39 ;:'错误&#39 ;,' c&#39 ;:' c&#39 ;:' c&#39 ;:' 39; 42P18&#39 ;,' m&#39 ;:'无法确定参数1美元&#39 ;,' f&#39 ;:' postgres.c&# 39 ;,' l&#39 ;:' ...&#39 ;,' r&#39 ;:' exec_parse_message'}

准备好的报表可用于提高性能时,当您有一个反复声明时会提高性能。这是一个例子:

>> >导入pg8000。天然>> > >> > con = pg8000。本国的。连接(" postgres"密码=" cpsnow")>> > >> > #创建准备好的声明>> > ps = con。准备("选择cast(:v作为varchar)")> > >> > #重复地兑换陈述>> > PS。运行(v ="速度")[['速度']]> > PS。跑(v =" rapt" ['']> > PS。跑(v =" swift")[[' swift']]> > >> > #关闭准备好的声明,释放服务器上的资源>> > PS。关闭()

>> >导入pg8000。天然>> >进口乐谱>> > >> > #使用当前用户名和gt连接到数据库;> >用户名= getpass。 getUser()>> >连接= PG8000。本国的。连接(用户名,密码=" cpsnow")>> > >> >联系。跑("选择' pilau'")[[' pilau']]]

>> >导入pg8000。天然>> >来自OS进口环境和GT;> > >> >用户名= Environ。得到(' pguser'' postgres')>> >密码=环境。得到(' pgpassword'' cpsnow')>> > Host = Environ。得到(' pghost'' localhost')>> >端口=环境。得到(' pgport&#39 ;,' 5432')>> >数据库=环境。得到(' pgdatabase')>> > >> >连接= PG8000。本国的。连接(...用户名,password = password,host = host,port = port,database = database)>> > >> >联系。跑("选择'开罗&#39先生;")[['开罗和#39先生;]]

它可能被问到,为什么PG8000没有这种行为?思考遵循Python Zen的第二个憎恶:

所以我们只采用了能够设置PG8000的连接参数的方法.Connection()构造函数。

可能是您的PostgreSQL服务器位于SSL代理服务器后面,可以在其中设置PG8000特定的属性SSL.SSLContext.Request_ssl = false,告诉pg8000使用anssl套接字连接,但不需要从PostgreSQL服务器请求SSL:

>> >导入pg8000。天然>> > >> > con = pg8000。本国的。连接(用户名,密码=" cpsnow")>> >骗子。跑("开始交易")[]>> >骗子。 run("声明c scrossolsor for select * from generate_series(1,100)")[]>> >骗子。跑("从c&#34上获取5.)[[1],[2],[3],[4],[5]]> >骗子。跑("从c&#34移动50;)[]>> >骗子。运行(" FETCH BACKWARD 10选自C")[[54],[53],[52],[51],[50],[49],[48],[47],[46 [45]]>> >骗子。跑("关闭c")[]>> >骗子。运行("回滚")[]

>> >导入pg8000。天然>> > >> > con = pg8000。本国的。连接(用户名,密码=" cpsnow")>> > >> > #创建一个blob并获得其OID>> >数据= B'你好' >> > res = con。运行("选择lo_from_bytea(0,:data)",data = data)>> > oid = res [0] [0]>> > >> > #创建一个表并存储blob&gt的oid;> >骗子。运行("创建临时表图像(光栅OID)")[]>> >骗子。运行("插入图像(栅格)值(:oid)",oid = oid)[]>> > #使用OID&gt检索数据。> >骗子。运行("选择lo_get(:oid)",oid = oid)[[b' hello']]>> > >> > #将一些数据添加到Blob&gt的末尾;> > more_data = B'所有' >> > offset = len(数据)>> >骗子。运行(..."选择lo_put(:oid,:offset,deam)",... oid = oid,offset = offset,data = more_data)[['&#39 ;]]]>> >骗子。 run("选择lo_get(:oid)",oid = oid)[[b'你好所有']>> > >> > #下载数据的一部分>> >骗子。运行("选择lo_get(:oid,6,3)",oid = oid)[[b'全部']]]

导入pg8000,连接数据库,创建一个表,添加一些行,然后添加一个表:

>> >导入pg8000。 dbapi>> > >> > conn = pg8000。 dbapi。 Connect(User =" Postgres",密码=" cpsnow")>> > Cursor = Conn。 Cursor()>> >光标。执行("创建临时表书(ID序列,标题文本)")>> >光标。执行(..."插入到书籍(标题)值(%s),(%s)返回id,title",...(" enter' s game" ,"死者的扬声器"))> >结果=光标。 fetchall()>> >在结果中为行:... ID,标题=行...打印(" ID =%s,title =%s"%(id,title))id = 1,标题= enter&#39 ; S游戏ID = 2,标题=死者的扬声器>> >康涅狄格州。犯罪()

>> >导入pg8000。 dbapi>> > >> > con = pg8000。 dbapi。 Connect(User =" Postgres",密码=" cpsnow")>> > Cursor = Con。 Cursor()>> > >> >光标。执行("从现在(从现在())")>> >光标。 fetchone()[3.0]

>> >导入DateTime>> >导入pg8000。 dbapi>> > >> > con = pg8000。 dbapi。连接(用户=" postgres"

......