python的flex服务端数据接口开发
python 如果给flex提供服务端，需要提供一个网关和一个可供客户端(flex)调用的类。这方面我更加推荐用twisted来写这个网关，因为twisted有很好的异步机制。
下面的我写的一个简单的验证用户的python服务端：
______________________________
DBServer.py
# Copyright (c) 2009-2010 The Newjh Project.
"""
@author: Roy
@since: 0.1.0
"""
import os.path
import ConfigParser
cfg = ConfigParser.SafeConfigParser()
cfg.read('settings.cfg')
from twisted.internet import reactor
from twisted.web import server as _server, static, resource
from twisted.enterprise import adbapi
from pyamf.remoting.gateway.twisted import TwistedGateway
from newjh import newjhService
root = resource.Resource()
gw = TwistedGateway({'newjh': newjhService(adbapi.ConnectionPool('MySQLdb',
host=cfg.get('db','host'), user=cfg.get('db','user'),
passwd=cfg.get('db','password'), db=cfg.get('db','database'),
cp_reconnect=True,use_unicode=True, charset='utf8'))}, expose_request=False)
root.putChild('gateway', gw)
root.putChild('crossdomain.xml', static.File(os.path.join(os.getcwd(),
os.path.dirname(__file__), 'crossdomain.xml'), defaultType='application/xml'))
server = _server.Site(root)
reactor.listenTCP(8000, server)
reactor.run()
————————————————
newjh.py
"""
Newjh remoting service.
@since: 0.1.0
"""
from datetime import datetime
from urlparse import urlparse
import re
from twisted.internet import defer
from twisted.internet.task import LoopingCall
import pyamf
from pyamf.flex import ArrayCollection, ObjectProxy
from pyamf.remoting.gateway import expose_request
EMAIL_RE = r"^.+@([?)[a-zA-Z0-9-.]+.([a-zA-Z]{2,3}|[0-9]{1,3})(]?)$"
# This is MySQL specific, make sure that if you use a different database server
# this is updated to ensure sql injection attacks don't occur
def sql_safe(value):
if isinstance(value, basestring):
return value.replace("'", "\'")
elif isinstance(type(value), (int, float)):
return value
raise TypeError, 'basestring, int or float expected'
def is_valid_url(url):
o = urlparse(url)
# scheme
if o[0] == '':
return (False, 'Scheme required')
if o[1] == '':
return (False, 'Hostname required')
return (True, None)
def is_valid_email(email):
"""
A very basic email address format validator
"""
if re.match(EMAIL_RE, email) != None:
return True
return False
def build_message(row):
m = User()
m.ID = row[0]
m.UserName = row[1]
m.Password = row[2]
return m
class User:
pass
pyamf.register_class(User, 'newjh.User')
class newjhService(object):
def __init__(self, pool):
self.conn_pool = pool
LoopingCall(self._keepAlive).start(3600, False)
def _keepAlive(self):
print 'Running Keep Alive...'
self.conn_pool.runOperation('SELECT 1')
def getUser(self, username, password):
"""
Gets all approved messages.
"""
print 'username : %s passwrd : %s' % (username, password)
def cb(rs):
print rs
ret = [ObjectProxy(build_message(row)) for row in rs]
print ret
return ArrayCollection(ret)
def eb(failure):
# TODO nick: logging
return ArrayCollection()
d = self.conn_pool.runQuery("SELECT * FROM user where " +
"UserName= '" +username + "' and Password = '" +
password +"'" ).addErrback(eb).addCallback(cb)
return d
——————————————————
crossdomain.xml
<?xml version="1.0"?>
<!DOCTYPE cross-domain-policy SYSTEM "http://www.macromedia.com/xml/dtds/cross-domain-policy.dtd">
<cross-domain-policy>
<allow-access-from domain="*" />
</cross-domain-policy>
——————————————————
settings.cfg
[db]
host: localhost
user: root
password: *********
database: newjh
————————————————————————
————————————————————————
flex 客户端：
客户端先建立联接，然后就可以向服务端call了，回调写在onResult
要注意的是gateway / newjh.getUser 和服务端要对应好
基类 extends NetConnection
connect("http://127.0.0.1:8000/gateway");
addEventListener(IOErrorEvent.IO_ERROR, ioErrorHandler);
addEventListener(NetStatusEvent.NET_STATUS, netStatusHandler);
responder = new Responder(onResult);
call(“newjh.getUser”, responder, event.data["name"], event.data["pass"]);
下面的我写的一个简单的验证用户的python服务端：
______________________________
DBServer.py
# Copyright (c) 2009-2010 The Newjh Project.
"""
@author: Roy
@since: 0.1.0
"""
import os.path
import ConfigParser
cfg = ConfigParser.SafeConfigParser()
cfg.read('settings.cfg')
from twisted.internet import reactor
from twisted.web import server as _server, static, resource
from twisted.enterprise import adbapi
from pyamf.remoting.gateway.twisted import TwistedGateway
from newjh import newjhService
root = resource.Resource()
gw = TwistedGateway({'newjh': newjhService(adbapi.ConnectionPool('MySQLdb',
host=cfg.get('db','host'), user=cfg.get('db','user'),
passwd=cfg.get('db','password'), db=cfg.get('db','database'),
cp_reconnect=True,use_unicode=True, charset='utf8'))}, expose_request=False)
root.putChild('gateway', gw)
root.putChild('crossdomain.xml', static.File(os.path.join(os.getcwd(),
os.path.dirname(__file__), 'crossdomain.xml'), defaultType='application/xml'))
server = _server.Site(root)
reactor.listenTCP(8000, server)
reactor.run()
————————————————
newjh.py
"""
Newjh remoting service.
@since: 0.1.0
"""
from datetime import datetime
from urlparse import urlparse
import re
from twisted.internet import defer
from twisted.internet.task import LoopingCall
import pyamf
from pyamf.flex import ArrayCollection, ObjectProxy
from pyamf.remoting.gateway import expose_request
EMAIL_RE = r"^.+@([?)[a-zA-Z0-9-.]+.([a-zA-Z]{2,3}|[0-9]{1,3})(]?)$"
# This is MySQL specific, make sure that if you use a different database server
# this is updated to ensure sql injection attacks don't occur
def sql_safe(value):
if isinstance(value, basestring):
return value.replace("'", "\'")
elif isinstance(type(value), (int, float)):
return value
raise TypeError, 'basestring, int or float expected'
def is_valid_url(url):
o = urlparse(url)
# scheme
if o[0] == '':
return (False, 'Scheme required')
if o[1] == '':
return (False, 'Hostname required')
return (True, None)
def is_valid_email(email):
"""
A very basic email address format validator
"""
if re.match(EMAIL_RE, email) != None:
return True
return False
def build_message(row):
m = User()
m.ID = row[0]
m.UserName = row[1]
m.Password = row[2]
return m
class User:
pass
pyamf.register_class(User, 'newjh.User')
class newjhService(object):
def __init__(self, pool):
self.conn_pool = pool
LoopingCall(self._keepAlive).start(3600, False)
def _keepAlive(self):
print 'Running Keep Alive...'
self.conn_pool.runOperation('SELECT 1')
def getUser(self, username, password):
"""
Gets all approved messages.
"""
print 'username : %s passwrd : %s' % (username, password)
def cb(rs):
print rs
ret = [ObjectProxy(build_message(row)) for row in rs]
print ret
return ArrayCollection(ret)
def eb(failure):
# TODO nick: logging
return ArrayCollection()
d = self.conn_pool.runQuery("SELECT * FROM user where " +
"UserName= '" +username + "' and Password = '" +
password +"'" ).addErrback(eb).addCallback(cb)
return d
——————————————————
crossdomain.xml
<?xml version="1.0"?>
<!DOCTYPE cross-domain-policy SYSTEM "http://www.macromedia.com/xml/dtds/cross-domain-policy.dtd">
<cross-domain-policy>
<allow-access-from domain="*" />
</cross-domain-policy>
——————————————————
settings.cfg
[db]
host: localhost
user: root
password: *********
database: newjh
————————————————————————
————————————————————————
flex 客户端：
客户端先建立联接，然后就可以向服务端call了，回调写在onResult
要注意的是gateway / newjh.getUser 和服务端要对应好
基类 extends NetConnection
connect("http://127.0.0.1:8000/gateway");
addEventListener(IOErrorEvent.IO_ERROR, ioErrorHandler);
addEventListener(NetStatusEvent.NET_STATUS, netStatusHandler);
responder = new Responder(onResult);
call(“newjh.getUser”, responder, event.data["name"], event.data["pass"]);