用于sql炼金术的线程安全、多会话/多连接模型存储库上下文库
alchemist-stack的Python项目详细描述
炼金术士堆栈
package author:h.d.“chip”麦卡洛四世
上次更新时间：2018年4月23日
说明：
一个灵活的模型库数据库栈，用于SQL Algimy
概述
炼金术士堆栈旨在成为线程安全、多会话/多连接
用法
ORM表格示例：
# table_example.pyfromalchemist_stack.repository.modelsimportBasefromsqlalchemyimportColumn,Integer,DateTimeclassExampleTable(Base):__tablename__='example'primary_key=Column('id',Integer,primary_key=True)timestamp=Column(DateTime(timezone=True),nullable=False)def__repr__(self):return'<Example(timestamp={timestamp})>'.format(timestamp=self.timestamp)
示例模型：
# model_example.pyfromtables.table_exampleimportExampleTablefromdatetimeimportdatetime,timezonefromtypingimportTypeVarE=TypeVar('E',bound="Example")classExample(object):""" Example Model class. """def__init__(self,timestamp:datetime=datetime.now(timezone.utc).astimezone(),primary_key:int=None,*args,**kwargs):self.__pk=primary_keyself.__timestamp=timestampself.__args=argsself.__kwargs=kwargsdef__call__(self,*args,**kwargs)->ExampleTable:""" Called when an instance of Example is called, e.g.: `x = Example(...)` `x(...)` This is equivalent to calling `to_orm()` on the object instance. :returns: The ORM of the Example. :rtype: ExampleTable """returnself.to_orm()def__repr__(self)->str:""" A detailed String representation of Example. :returns: String representation of Example object. """return'<class Test(pk={pk} timestamp={timestamp}) at {hex_id}>'.format(pk=self.__pk,timestamp=self.__timestamp,hex_id=hex(id(self)))@propertydefid(self)->int:returnself.__pk@propertydeftimestamp(self)->datetime:returnself.__timestampdefto_orm(self)->ExampleTable:returnExampleTable(primary_key=self.__pk,timestamp=self.__timestamp)@classmethoddeffrom_orm(cls,obj:ExampleTable)->E:returncls(timestamp=obj.timestamp,primary_key=obj.primary_key)
存储库示例：
# repository_example.pyfromalchemist_stack.contextimportContextfromalchemist_stack.repositoryimportRepositoryBasefrommodels.model_exampleimportExamplefromtables.table_exampleimportExampleTableclassExampleRepository(RepositoryBase):""" """def__init__(self,context:Context,*args,**kwargs):""" Test Repository Constructor :param database: The Database object containing the engine used to connect to the database. :param args: Additional Arguments :param kwargs: Additional Keyword Arguments """super().__init__(context=context,*args,**kwargs)def__repr__(self):""" :return: """return'<class ExampleRepository->RepositoryBase(context={context}) at {hex_id}>'\ .format(context=str(self.context),hex_id=hex(id(self.context)))defcreate_example(self,obj:Example):self._create_object(obj=obj.to_orm())defget_example_by_id(self,example_id:int)->Example:self._create_session()__query=self._read_object(cls=ExampleTable)__t=__query.with_session(self.session).get(example_id)self._close_session()ifisinstance(__t,ExampleTable):returnExample.from_orm(__t)defupdate_example_by_id(self,example_id:int,values:dict,synchronize_session:str='evaluate')->int:self._create_session()__query=self._update_object(cls=ExampleTable,values=values)rowcount=__query.with_session(self.session)\ .filter(ExampleTable.primary_key==example_id)\ .update(values=values,synchronize_session=synchronize_session)self._commit_session()returnrowcount
欢迎加入QQ群-->： 979659372
