pycurl urllib fallback
alex goretoy
aleksandr.goretoy at gmail.com
Tue Dec 30 22:39:23 EST 2008
Hello All,
I have this class I like to call pcrunchly. That works to do all my request
via libcurl library. What I want to do is add capability for this class to
fallback to urllib if pycurl module is not install or is not importable for
some reason. I'm posting my whole class. Use it however you want, just
please help me to make it better. As you can see from the code I've already
started to add this functionality. I found some code on Google that I'm
trying to combine with my class, so that it can achieve this task. I want to
ask the reader of this thread. Is this a good conservative thing to do. Will
this make my class to bulky. Is this something I want? or do I want
something else? Is there a better way to achieve this perhaps? I would
appreciate any comments on this matter. I would like this class to be able
to handle any protocol you throw at it. So as to make it other apps that
need to use a different protocol and stuff. Please notice the first huge
comment. That is what I'm currently combining into my working pycurl class.
Is this good? Oh yeah, please don't mind all those self.soc functions calls.
They print things in color to my terminal for me.
#!/usr/bin/env python
from ctypes import *
import os, sys, types, urllib, urllib2, urlparse
import stdout_colours
#1. #!/usr/bin/env python
#2. # -*- coding: UTF-8 -*-
#3.
#4. import cookielib
#5. import urllib
#6. import urllib2
#7.
#8. cj = cookielib.CookieJar()
#9. opener = urllib2.build_opener(urllib2.HTTPCookieProcessor(c j))
#10. resp = opener.open('http://www.amm.com/login.asp') # save a cookie
#11.
#12. theurl = 'http://www.amm.com/login.asp'
#13. # an example url that sets a cookie, try different urls here and see
the cookie collection you can make !
#14. body={'username':'AMMT54590570','password':'AMMT32 564288'}
#15. txdata = urllib.urlencode(body)
#16. # if we were making a POST type request, we could encode a dictionary
of values here - using urllib.urlencode
#17. txheaders = {'User-agent' : 'Mozilla/4.0 (compatible; MSIE 5.5;
Windows NT)'}
#18. # fake a user agent, some websites (like google) don't like automated
exploration
#19.
#20.
#21. try:
#22. req = urllib2.Request(theurl, txdata, txheaders) # create a
request object
#23. handle = opener.open(req) # and open it to return a handle on the
url
#24. HTMLSource = handle.read()
#25. f = file('test.html', 'w')
#26. f.write(HTMLSource)
#27. f.close()
#28.
#29. except IOError, e:
#30. print 'We failed to open "%s".' % theurl
#31. if hasattr(e, 'code'):
#32. print 'We failed with error code - %s.' % e.code
#33. elif hasattr(e, 'reason'):
#34. print "The error object has the following 'reason' attribute
:", e.reason
#35. print "This usually means the server doesn't exist, is down,
or we don't have an internet connection."
#36. sys.exit()
#37.
#38. else:
#39. print 'Here are the headers of the page :'
#40. print handle.info() # handle.read() returns the page,
handle.geturl() returns the true url of the page fetched (in case urlopen
has followed any redirects, which it sometimes does)
class curl(object):
"Encapsulate user operations on CGIs through curl."
def __init__(self, base_url=""):
self.func_me_color="white_on_black"
self.soc=stdout_colours.stdout_colors()
self.soc.me_him(['ENTER:',__name__],self.func_me_color)
# These members might be set.
self.base_url = base_url
self.verbose = 0
self.response = ""
self.PYCURL=None
try:
import pycurl
self.PYCURL = True
except ImportError,e:
import cookielib
import urllib
import urllib2
self.PYCURL = False
#print "\ntrouble with importing pycurl %s\n" % e
#sys.exit(1)
# Nothing past here should be modified by the caller.
if self.PYCURL:
self.curlobj = pycurl.Curl()
# Verify that we've got the right site...
#self.curlobj.setopt(pycurl.SSL_VERIFYHOST, 2)
# Follow redirects in case it wants to take us to a CGI...
self.curlobj.setopt(pycurl.FOLLOWLOCATION, 1)
#self.curlobj.setopt(pycurl.MAXREDIRS, 15)
# Setting this option with even a nonexistent file makes libcurl
# handle cookie capture and playback automatically.
self.curlobj.setopt(pycurl.COOKIEFILE, "/dev/null")
# Set timeouts to avoid hanging too long
self.curlobj.setopt(pycurl.CONNECTTIMEOUT, 60)
self.curlobj.setopt(pycurl.TIMEOUT, 600)
#self.set_verbosity(self.verbose)
# Set up a callback to capture
else:
self.cj=cookielib.CookieJar()
print "\nurllib:\n"
def response_callback(x):
self.soc.me_him(['ENTER:',__name__],self.func_me_color)
if self.PYCURL:
self.response += x
self.curlobj.setopt(pycurl.WRITEFUNCTION, response_callback)
self.soc.me_him(['EXIT:',__name__],self.func_me_color)
def set_verbosity(self, level):
"Set verbosity to 1 to see transactions."
self.soc.me_him(['ENTER:',__name__],self.func_me_color)
if self.PYCURL:
self.verbose=int(level)
self.curlobj.setopt(pycurl.VERBOSE, self.verbose)
self.soc.me_him(['EXIT:',__name__],self.func_me_color)
def get(self, cgi, params=""):
"Ship a GET request to a specified CGI, capture the response body."
self.soc.me_him(['ENTER:',__name__],self.func_me_color)
if self.PYCURL:
if params:
cgi += "?" + urllib.urlencode(params)
self.curlobj.setopt(pycurl.URL, os.path.join(self.base_url,
cgi))
self.curlobj.setopt(pycurl.HTTPGET, 1)
self.response = ""
self.curlobj.perform()
# if self.verbose > 0:
# print self.response
self.soc.me_him(['EXIT:',__name__],self.func_me_color)
def post(self, cgi, params):
"Ship a POST request to a specified CGI, capture the response
body.."
self.soc.me_him(['ENTER:',__name__],self.func_me_color)
if self.PYCURL:
self.curlobj.setopt(pycurl.URL, os.path.join(self.base_url,
cgi))
self.curlobj.setopt(pycurl.POST, 1)
self.curlobj.setopt(pycurl.POSTFIELDS, urllib.urlencode(params))
self.response = ""
self.curlobj.perform()
# if self.verbose>0:
# print self.response
self.soc.me_him(['EXIT:',__name__],self.func_me_color)
def upload(self, cgi, file_name, file):
"POST file from localhost to location/cgi."
self.soc.me_him(['ENTER:',__name__],self.func_me_color)
if self.PYCURL:
self.curlobj.setopt(pycurl.URL, os.path.join(self.base_url,
cgi))
self.curlobj.setopt(pycurl.HTTPPOST,[(file_name,
(pycurl.FORM_FILE,file))])
self.response = ""
self.curlobj.perform()
if self.verbose>0:
print self.response
self.soc.me_him(['EXIT:',__name__],self.func_me_color)
# --------------------------------
# DJANGO RECEIVE TEST APPLICATION
# --------------------------------
# --------- urls.py ----------------
#from django.conf.urls.defaults import *
#urlpatterns = patterns('',
#(r'^receive/$', 'web.views.receive'),
#)
# --------- web\views.py ----------------
#def receive(request):
#assert request.method=="POST"
#print "receive.META.SERVER_PORT", request.META["SERVER_PORT"],
request.POST
#files = []
#for multipart_name in request.FILES.keys():
#multipart_obj = request.FILES[multipart_name]
#content_type = multipart_obj['content-type']
#filename = multipart_obj['filename']
#content = multipart_obj['content']
#files.append((filename, content_type, content))
#import datetime
# write file to the system - add timestamp in the name
#file("c:\\tmp\\%s_%s" %
(datetime.datetime.now().isoformat().replace(":", "-"), filename),
"wb").write(content)
#fnames = ",".join([fname for fname, ct, c in files])
#return HttpResponse("me-%s-RECEIVE-OK[POST=%s,files=%s]" %
(request.META["SERVER_PORT"], request.POST.values(), fnames ))
def answered(self, check):
"Does a given check string occur in the response?"
self.soc.me_him(['ENTER:',__name__],self.func_me_color)
self.soc.me_him(['RETURN:',__name__],self.func_me_color)
if self.PYCURL:
return self.response.find(check) >= 0
def close(self):
"Close a session, freeing resources."
self.soc.me_him(['ENTER:',__name__],self.func_me_color)
if self.PYCURL:
self.curlobj.close()
self.soc.me_him(['EXIT:',__name__],self.func_me_color)
class session(curl):
def login(self, cgisite,username, password):
"""login -
cgi="login.php",params=(("username",name),("password",pass),("foo","bar"))
"""
self.soc.me_him(['ENTER:',__name__],self.func_me_color)
self.post(cgisite, (("username",username),
("password",password)))
self.soc.me_him(['EXIT:',__name__],self.func_me_color)
def logout(self, cgisite):
"""logout - cgi="logout.php" """
self.soc.me_him(['ENTER:',__name__],self.func_me_color)
self.get(cgisite)
self.soc.me_him(['EXIT:',__name__],self.func_me_color)
if __name__ == "__main__":
if len(sys.argv) < 3:
print "Usage: %s \"schema://site/cgi\" \"username\" \"password\"" %
sys.argv[0]
site=sys.argv[1]
username=sys.argv[2]
password=sys.argv[3]
sess=session("")
sess.set_verbosity(1)
sess.login(site,username,password)
a=""
for i in range(len(password)):
a+="*"
print "YOU ARE LOGGED IN!",site,username,a
sess.logout()
sess.close()
--
А-Б-В-Г-Д-Е-Ё-Ж-З-И-Й-К-Л-М-Н-О-П-Р-С-Т-У-Ф-Х-Ц-Ч-Ш-Щ-Ъ-Ы-Ь-Э-Ю-Я
а-б-в-г-д-е-ё-ж-з-и-й-к-л-м-н-о-п-р-с-т-у-ф-х-ц-ч-ш-щ-ъ-ы-ь-э-ю-я
-------------- next part --------------
An HTML attachment was scrubbed...
URL: <http://mail.python.org/pipermail/python-list/attachments/20081231/5e96f529/attachment-0001.html>
More information about the Python-list
mailing list