一个小巧的MySQL Shell-数据库专栏,MySQL-数据库其他-idc中文资讯站-客观公证的IDC产业权威媒体,关注域名、虚拟主机、服务器、网络、开源技术资讯，整合运营商资源，客观评价各地机房带宽线路质量、IDC企业名录等信息

一个小巧的MySQL Shell-数据库专栏,MySQL

来源：作者：发布时间:2007-12-25 13:47:41

#!/usr/bin/python
# -*- coding: cp936 -*-

#################################
#   written by caocao           #
#   caocao@eastday.com          #
#   http://nethermit.yeah.net   #
#################################

import sys, string, os
from types import *
import mysqldb

print "written by caocao"
print "caocao@eastday.com"
print "http://nethermit.yeah.net"
print

def iif(expression, whentrue, whenfalse):
if expression:
  return whentrue
else:
  return whenfalse

class mysqltest:
def __init__(self, host="localhost", user="root", passwd="", db=""):
  self.connection=none
  self.host=host
  self.user=user
  self.passwd=passwd
  self.db=db
  self.result=none

  print "-"*40
  print "mysql shell v 1.0"
  print "usage: python mysql.shell.py [host] [user] [passwd(% is empty)] [db]"
  print "connect..."
  try:
   self.connection=mysqldb.connect(host=self.host, user=self.user, passwd=self.passwd, db=self.db)
  except:
   print "cant connect to mysql server.＼nplease make sure your username or password is right."
   sys.exit(1)
  print "-"*40
  print self.printcomment("connection", "get_server_info")
  print self.printcomment("connection", "get_host_info")
  print self.printcomment("connection", "get_proto_info")
  print self.printcomment("connection", "info")
  print self.printcomment("connection", "character_set_name")
  print self.printcomment("connection", "thread_id")
  print self.printcomment("connection", "stat")

def __del__(self):
  if self.connection!=none:
   self.connection.close()
  print "-"*40
  print "quit..."

def printcomment(self, instance, function):
  return "%s = %s" % (string.rjust(function, 18), eval("self."+instance+"."+function+"()"))

def printall(self):
  output, row="", self.result.fetch_row(0)
  for i in range(self.result.num_fields()):
   output+=repr(self.result.describe()[i][0])+"＼n"
  for i in range(self.result.num_rows()):
   for j in range(self.result.num_fields()):
    output+=iif(type(row[i][j]) is stringtype, row[i][j], repr(row[i][j]))+"＼n"
  return output

def runsql(self, querystring="show databases"):
  print "-"*40
  try:
   self.connection.query(querystring)
  except:
   print "cant run sql."
  else:
   self.result=self.connection.store_result()
   print self.printcomment("connection", "field_count")
   print self.printcomment("connection", "affected_rows")
   print self.printcomment("connection", "insert_id")
   print self.printcomment("result", "num_fields")
   print self.printcomment("result", "num_rows")
   print self.printcomment("result", "field_flags")
   print "-"*40
   print self.printall()

if __name__=="__main__":
argarray=sys.argv
del argarray[0]
test=eval(("mysqltest(＼"%s＼")" % "＼",＼"".join(argarray)).replace("%", ""))
while true:
  try:
   command=string.strip(raw_input("ps mysql>"), " ")
   commandlow=string.lower(command)
  except eoferror:
   break
  else:
   if commandlow=="exit" or commandlow=="quit":
    break
   elif commandlow=="":
    continue
   else:
    test.runsql(command)

喜欢本文，那就收藏到：