ä¸ãåè¨
1 ãæå¨å®ä¾åä¸ä¸ªuser对象çæ¶åï¼å¯ä»¥user=Userï¼name='lqz',password='123'ï¼
2 ãä¹å¯ä»¥ user=User()
ããããuser['name']='lqz'
ããããuser['password']='123'
3 ãä¹å¯ä»¥ user=User()
ããããuser.name='lqz'
ããããuser.password='password'
å两ç§ï¼å¯ä»¥éè¿ç»§æ¿åå ¸dictæ¥å®ç°ï¼ç¬¬ä¸ç§ï¼ç¨getattråsetattr
   __getattr__ æ¦æªç¹å·è¿ç®ãå½å¯¹æªå®ä¹çå±æ§å称åå®ä¾è¿è¡ç¹å·è¿ç®æ¶ï¼å°±ä¼ç¨å±æ§åä½ä¸ºå符串è°ç¨è¿ä¸ªæ¹æ³ãå¦æ继æ¿æ å¯ä»¥æ¾å°è¯¥å±æ§ï¼åä¸è°ç¨æ¤æ¹æ³
   __setattr__ä¼æ¦æªææå±æ§ççèµå¼è¯å¥ãå¦æå®ä¹äºè¿ä¸ªæ¹æ³ï¼self.arrt = value å°±ä¼åæself,__setattr__("attr", value).è¿ä¸ªéè¦æ³¨æãå½å¨__setattr__æ¹æ³å 对å±æ§è¿è¡èµå¼æ¯ï¼ä¸å¯ä½¿ç¨self.attr = value,å 为ä»ä¼å次è°ç¨self,__setattr__("attr", value),åä¼å½¢ææ ç©·éå½å¾ªç¯ï¼æå导è´å æ 溢åºå¼å¸¸ãåºè¯¥éè¿å¯¹å±æ§åå ¸åç´¢å¼è¿ç®æ¥èµå¼ä»»ä½å®ä¾å±æ§ï¼ä¹å°±æ¯ä½¿ç¨self.__dict__['name'] = value
äºãå®ä¹Field
æ°æ®åºä¸æ¯ä¸åæ°æ®ï¼é½æï¼ååï¼åçæ°æ®ç±»åï¼æ¯å¦æ¯ä¸»é®ï¼é»è®¤å¼
class Field:
def __init__(self, name, column_type, primary_key, default_value):
self.name = name
self.column_type = column_type
self.primary_key = primary_key
self.default_value = default_value
class StringField(Field):
def __init__(self, name, column_type='varchar(100)', primary_key=False, default_value=None):
super().__init__(name, column_type, primary_key, default_value)
class IntegerField(Field):
def __init__(self, name, primary_key=False, default_value=0):
super().__init__(name, 'int', primary_key, default_value)
ä¸ãå®ä¹å ç±»
æ°æ®åºä¸çæ¯ä¸ªè¡¨ï¼é½æ表åï¼æ¯ä¸åçååï¼ä»¥å主é®æ¯åªä¸å
æ¢ç¶æè¦ç¨æ°æ®åºä¸ç表ï¼å¯¹åºè¿ä¸ä¸ªç¨åºä¸çç±»ï¼é£ä¹æè¿ä¸ªç±»ä¹åºè¯¥æè¿äºç±»å±æ§
ä½æ¯ä¸åçç±»è¿äºç±»å±æ§åä¸å°½ç¸åï¼æ以æåºè¯¥æä¹åï¼å¨å ç±»éæ¦æªç±»çå建è¿ç¨ï¼ç¶åæè¿äºä¸è¥¿ååºæ¥ï¼æ¾å°ç±»éé¢
æ以ç¨å°äºå ç±»
class ModelMetaclass(type): def __new__(cls, name, bases,attrs): if name=='Model': return type.__new__(cls,name,bases,attrs) table_name=attrs.get('table_name',None) if not table: table_name=name primary_key=None mappings=dict() for k,v in attrs.items(): if isinstance(v,Field):#v æ¯ä¸æ¯Fieldç对象 mappings[k]=v if v.primary_key: #æ¾å°ä¸»é® if primary_key: raise TypeError('主é®éå¤ï¼%s'%k) primary_key=k for k in mappings.keys(): attrs.pop(k) if not primary_key: raise TypeError('没æ主é®') attrs['table_name']=table_name attrs['primary_key']=primary_key attrs['mappings']=mappingsreturn         type.__new__(cls,name,bases,attrs)
åãå®ä¹Modelåºç±»
Modelç±»æ¯ææè¦å¯¹åºæ°æ®åºè¡¨ç±»çåºç±»ï¼æ¯ä¸ªæ°æ®åºè¡¨å¯¹åºç±»ç对象ï¼é½åºè¯¥ææ¥è¯¢ï¼æå ¥ï¼ä¿åï¼æ¹æ³
éè¿ç»§æ¿åå ¸dictæ¥å®ç°å®ä¾åï¼å¹¶éågetattråsetattr
class Model(dict, metaclass=ModelMetaclass): def __init__(self, **kw): super(Model, self).__init__(**kw) def __getattr__(self, key): # .访é®å±æ§è§¦å try: return self[key] except KeyError: raise AttributeError('没æå±æ§ï¼%s' % key) def __setattr__(self, key, value): self[key] = value @classmethod def select_all(cls, **kwargs): ms = mysql_singleton.Mysql().singleton() if kwargs: # å½æåæ°ä¼ å ¥çæ¶å key = list(kwargs.keys())[0] value = kwargs[key] sql = "select * from %s where %s=?" % (cls.table_name, key) sql = sql.replace('?', '%s') re = ms.select(sql, value) else: # å½æ åä¼ å ¥çæ¶åæ¥è¯¢ææ sql = "select * from %s" % cls.table_name re = ms.select(sql) return [cls(**r) for r in re] @classmethod def select_one(cls, **kwargs): # æ¤å¤åªæ¯æåä¸æ¡ä»¶æ¥è¯¢ key = list(kwargs.keys())[0] value = kwargs[key] ms = mysql_singleton.Mysql().singleton() sql = "select * from %s where %s=?" % (cls.table_name, key) sql = sql.replace('?', '%s') re = ms.select(sql, value) if re: return cls(**re[0]) else: return None def save(self): ms = mysql_singleton.Mysql().singleton() fields = [] params = [] args = [] for k, v in self.mapping.items(): fields.append(v.name) params.append('?') args.append(getattr(self, k, v.default)) sql = "insert into %s (%s) values (%s)" % (self.table_name, ','.join(fields), ','.join(params)) sql = sql.replace('?', '%s') ms.execute(sql, args) def update(self): ms = mysql_singleton.Mysql().singleton() fields = [] args = [] pr = None for k, v in self.mapping.items(): if v.primary_key: pr = getattr(self, k, v.default) else: fields.append(v.name + '=?') args.append(getattr(self, k, v.default)) sql = "update %s set %s where %s = %s" % ( self.table_name, ', '.join(fields), self.primary_key, pr) sql = sql.replace('?', '%s') print(sql) ms.execute(sql, args)
äºãåºäºpymsqlçæ°æ®åºæä½ç±»ï¼åä¾ï¼
from conf import setting import pymysql class Mysql: __instance = None def __init__(self): self.conn = pymysql.connect(host=setting.host, user=setting.user, password=setting.password, database=setting.database, charset=setting.charset, autocommit=setting.autocommit) self.cursor = self.conn.cursor(cursor=pymysql.cursors.DictCursor) def close_db(self): self.conn.close() def select(self, sql, args=None): self.cursor.execute(sql, args) rs = self.cursor.fetchall() return rs def execute(self, sql, args): try: self.cursor.execute(sql, args) affected = self.cursor.rowcount # self.conn.commit() except BaseException as e: print(e) return affected @classmethod def singleton(cls): if not cls.__instance: cls.__instance = cls() return cls.__instance
å ãåºäºæ°æ®åºè¿æ¥æ± å®ç°çæ°æ®åºæä½ç±»
        db_pool.py
import pymysql from conf import setting from DBUtils.PooledDB import PooledDB POOL = PooledDB( creator=pymysql, # 使ç¨é¾æ¥æ°æ®åºç模å maxconnections=6, # è¿æ¥æ± å 许çæ大è¿æ¥æ°ï¼0åNone表示ä¸éå¶è¿æ¥æ° mincached=6, # åå§åæ¶ï¼é¾æ¥æ± ä¸è³å°å建ç空é²çé¾æ¥ï¼0表示ä¸å建 maxcached=5, # é¾æ¥æ± ä¸æå¤é²ç½®çé¾æ¥ï¼0åNoneä¸éå¶ maxshared=3, # é¾æ¥æ± ä¸æå¤å ±äº«çé¾æ¥æ°éï¼0åNoneè¡¨ç¤ºå ¨é¨å ±äº«ã blocking=True, # è¿æ¥æ± ä¸å¦æ没æå¯ç¨è¿æ¥åï¼æ¯å¦é»å¡çå¾ ãTrueï¼çå¾ ï¼Falseï¼ä¸çå¾ ç¶åæ¥é maxusage=None, # ä¸ä¸ªé¾æ¥æå¤è¢«éå¤ä½¿ç¨ç次æ°ï¼None表示æ éå¶ setsession=[], # å¼å§ä¼è¯åæ§è¡çå½ä»¤å表ã ping=0, # ping MySQLæå¡ç«¯ï¼æ£æ¥æ¯å¦æå¡å¯ç¨ã host=setting.host, port=setting.port, user=setting.user, password=setting.password, database=setting.database, charset=setting.charset, autocommit=setting.autocommit )
mysql_pool.py
setting.pyimport pymysql from ormpool import db_pool from threading import current_thread class MysqlPool: def __init__(self): self.conn = db_pool.POOL.connection() # print(db_pool.POOL) # print(current_thread().getName(), 'æ¿å°è¿æ¥', self.conn) # print(current_thread().getName(), 'æ± åéç®åæ', db_pool.POOL._idle_cache, '\r\n') self.cursor = self.conn.cursor(cursor=pymysql.cursors.DictCursor) def close_db(self): self.cursor.close() self.conn.close() def select(self, sql, args=None): self.cursor.execute(sql, args) rs = self.cursor.fetchall() return rs def execute(self, sql, args): try: self.cursor.execute(sql, args) affected = self.cursor.rowcount # self.conn.commit() except BaseException as e: print(e) finally: self.close_db() return affected
host = '127.0.0.1' port = 3306 user = 'root' password = '123456' database = 'youku2' charset = 'utf8' autocommit = True