在线时间:8:00-16:00
迪恩网络APP
随时随地掌握行业动态
扫描二维码
关注迪恩网络微信公众号
开源软件名称:easy_tormysql开源软件地址:https://gitee.com/NightRavenReady/easy_tormysql开源软件介绍:使用本轮子可以在项目中更方便的执行异步的MySQL数据操作,本轮子是基于TorMySql的二次封装,借鉴了Django models的编码风格。 支持 tornado 和 asyncio. python3 python2 初始化连接池from easy_tormysql import init_mysqlinit_mysql( default={ "max_connections" : 20, #max open connections "idle_seconds" : 7200, #conntion idle timeout time, 0 is not timeout "wait_connection_timeout" : 3, #wait connection timeout "host": "127.0.0.1", "user": "root", "passwd": "root", "charset": "utf8", "db": "example1" }, other_connection={ "host": "127.0.0.1", "user": "root", "passwd": "root", "charset": "utf8", "db": "example2", """ 系统会在默认数据库中查询表,如果你的表不在默认数据库中,请在other_connection的tables中进行指定 """ "tables": [ "example_table_mapping_class_lowercase_name" ] }) 定义 models
from easy_tormysql import BaseModel, Fieldclass Subscriber(BaseModel): """ 如果不指定db_table,系统会使用model类名的小写作为表名 """ db_table = 'tb_user' name = Field() create_time = Field(auto_now_add=True) login_time = Field(auto_now=True)
from easy_tormysql import BaseModel, Field, ForeignKey, ManyToManyFieldclass Author(BaseModel): name = Field()class Tag(BaseModel): name = Field()class Article(BaseModel): content = Field() create_time = Field() # 一对多关系 author = ForeignKey(Author) # 多对多关系:需要定义中间表 tags = ManyToManyField(Tag, middle_table='article_tags') 在 tornado 中使用
# 单表author = Author(name='Wang')# 一对多article = Article(content='My story...')author.article_set.add(article)yield author.save()# 多对多tag1, tag2 = Tag(name='poetry'), Tag(name='biography')yield tag1.save()yield tag2.save()article.tags.add(tag1)article.tags.add(tag2)yield article.save()
# 查询唯一记录author = yield Author.get(name='Wang')# 查询所有记录articles = yield Article.all()# 条件查询authors = yield Author.filter(name='Wang')authors = yield Author.filter(name__in=('Wang','Lee'))authors = yield Author.filter(name__contains='W')articles = yield Author.filter(create_time__between=(date1,date2))# 排序sorted_authors = yield Author.all(order_fields=("name",))# 分组records = yield Author.filter(name='Wang', group_fields=("name",))# 一对多articles = yield author.article_set.all()# 多对多tags = yield article.tags.all()articles = yield tag1.articles.all()
article.content = "programming..."yield article.save()
article.tags.remove(tag1)article.tags.remove(tag2)yield article.save()yield article.delete()
from tornado.web import RequestHandlerfrom tornado.gen import coroutineclass ExampleHandler(RequestHandler): @coroutine def get(self): ... # 单表 author = Author(name='Wang') # 一对多 article = Article(content='My story...') author.article_set.add(article) yield author.save() ... 在 asyncio 中使用
# 单表author = Author(name='Wang')# 一对多article = Article(content='My story...')author.article_set.add(article)await author.save()# 多对多tag1, tag2 = Tag(name='poetry'), Tag(name='biography')await tag1.save()await tag2.save()article.tags.add(tag1)article.tags.add(tag2)await article.save()
# 查询唯一记录author = await Author.get(name='Wang')# 查询所有记录articles = await Article.all()# 条件查询authors = await Author.filter(name='Wang')authors = await Author.filter(name__in=('Wang','Lee'))authors = await Author.filter(name__contains='W')articles = await Author.filter(create_time__between=(date1,date2))# 排序sorted_authors = await Author.all(order_fields=("name",))# 分组records = await Author.filter(name='Wang', group_fields=("name",))# 一对多articles = await author.article_set.all()# 多对多tags = await article.tags.all()articles = await tag1.articles.all()
article.content = "programming..."await article.save()
article.tags.remove(tag1)article.tags.remove(tag2)await article.save()await article.delete()
import asyncioasync def example(): ... # single author = Author(name='Wang') # one-to-many article = Article(content='My story...') author.article_set.add(article) await author.save() ... ioloop = asyncio.events.get_event_loop()ioloop.run_until_complete(example()) |
请发表评论