Async rest api python

Создание REST API с использованием aiohttp framework python

Асинхронное программирование приложений — последнее слово в мире разработки приложений. Оно постепенно набирает обороты и популярность среди предприятий и разработчиков приложений благодаря самой природе того, что асинхронные приложения могут выполнять несколько единиц работы одновременно, что может значительно сократить повысить скорость отклика приложения.

В последнее время больше внимания уделяется улучшению взаимодействия с пользователем, созданию приложений, ориентированных на бизнес-возможности и облачные технологии, способствующим адаптации микросервисов и бессерверной архитектуры, переопределению способов создания приложений и асинхронных приложений. стать путем будущего.

Интерфейсы прикладного программирования (API) стандартизировали способ взаимодействия прикладных программ друг с другом независимо от их реализации, и теперь еще труднее представить какое-либо микросервисное приложение, которое можно создать без использования API.

В этой статье мы будем создавать простой асинхронный REST API на python с использованием фреймворка aiohttp. Aiohttp — один из популярных асинхронных фреймворков, который использует неблокирующие сокеты и питается от библиотеки asyncio Python. Помимо использования в качестве веб-фреймворка сервера, его также можно использовать в качестве клиентского фреймворка.

В случае, если вы все еще задаетесь вопросом, что такое асинхронное программирование, я написал еще одну статью с подробным описанием как синхронных, так и асинхронных концепций, то же самое можно прочитать здесь. Не думая о будущем, давайте начнем с аспектов реализации.

REST API — реализации

В этом разделе мы будем реализовывать следующие конечные точки REST API с использованием асинхронной инфраструктуры aiohttp.

• /health использует метод GET — отображение статуса приложения.
• /v1/crypto/info использует метод POST — для получения информации о криптовалюте с помощью API от coin gecko.

Фактический код, который реализует вышеуказанные конечные точки с использованием фреймворков aiohttp, показан ниже.

Изучая приведенную выше реализацию, мы могли убедиться, что смогли запустить и запустить API с помощью всего нескольких строк кода (точнее, 28 строк), и это действительно демонстрирует скорость и масштаб, с которыми эти API могут быть построены и масштабированы. .

Если вы могли заметить, мы использовали ключевое слово async перед определением функции, и это указывает python, что эта функция является асинхронным методом и может выполняться асинхронно. Всю реализацию можно разбить на следующие три основные функции.

async def init() — инициализирует конечные точки приложения и службы.

Это функция инициализации, которая используется для инициализации нашего веб-приложения и определения двух конечных точек с помощью router.add_get для реализации метода GET и router.add_post. для реализации метода POST.

Конечные точки/v1/crypto/infoи/health и соответствующие им функции get_crypto_info и работоспособность, которые должны выполняться при доступе к конечным точкам, также определяются в этой функции инициализации.

async get_crypto_info(request) — обрабатывает запрос API.

Эта функция содержит фактическую логику реализации конечной точки (/v1/crypto/info), где мы обращаемся к API, предоставленному coin gecko для получения информации о криптовалюте. , биткойн в этом случае и детали возвращаются потребителю в формате json.

Возможно, мы должны были реализовать это как метод GET, но мы намеренно решили реализовать его как метод POST, просто чтобы продемонстрировать, как мы можем реализовать различные HTTP-глаголы (GET, POST, PUT, DELETE и т. д.) с использованием этих асинхронных фреймворков.

async def health(request) — проверка работоспособности.

Эта функция является фактической реализацией конечных точек /health, которая возвращает состояние работоспособности приложения.

Результаты теста

Мы использовали postman для тестирования этих конечных точек API, и результаты показаны ниже. На приведенном ниже снимке экрана показан результат выполнения «/v1/crypto/info»с использованием метода POST.

На приведенном ниже снимке экрана показан результат выполнения «/health»с использованием метода GET.

Сводка

API произвел революцию в том, как компьютерные приложения взаимодействуют друг с другом, а Async, с другой стороны, вырос как потенциальный инструмент для создания параллельных приложений и, объединив эти две технологии, предоставляет нам бесконечные возможности и возможности, ограниченные только нашим воображением.

Асинхронно реализованный API не только выполняется параллельно, но также расширяет возможности параллелизма для своих потребителей, а также значительно сокращает время отклика.

При этом, как и любые другие фреймворки, асинхронность имеет свои недостатки и может привести к снижению производительности в зависимости от характера приложения, а асинхронность также может изменить правила игры, если ее использовать в правильных случаях.

Надеюсь, мы вдохновили вас начать создавать свои приложения асинхронным способом.

До следующего раза… Приятного чтения..

Ссылка

Источник

How to create REST API using aiohttp

This provides concurrency especially for I/O bound tasks over sockets and other resources. Concurrency ensures that user does not have wait for the I/O bound results.

In this article, we will create a rest API for our application using aiohttp. It is a simple application which has a Note table.

Set up aiohttp

Activate a virtual environment in python 3 and install aiohttp

or clone the github repository and install the requirements

pip install -r requirements.txt 

Create Models

We will configure application to use sqlite as our database in models.py

# DBSession = scoped_session(sessionmaker(extension=ZopeTransactionExtension())) DB_URI = 'sqlite:///stuff.db' Session = sessionmaker(autocommit=False, autoflush=False, bind=create_engine(DB_URI)) session = scoped_session(Session) Base = declarative_base() 

Then we create Note class for note objects in models.py

class Note(Base): __tablename__ = 'notes' id = Column(Integer, primary_key=True) title = Column(String(50)) description = Column(String(50)) created_at = Column(String(50)) created_by = Column(String(50)) priority = Column(Integer) def __init__(self, title, description, created_at ,created_by, priority): self.title = title self.description = description self.created_at = created_at self.created_by = created_by self.priority = priority @classmethod def from_json(cls, data): return cls(**data) def to_json(self): to_serialize = ['id', 'title', 'description', 'created_at', 'created_by', 'priority'] d = <> for attr_name in to_serialize: d[attr_name] = getattr(self, attr_name) return d 

Resources

We define our API endpoints in aiohttp_rest.py file.

DEFAULT_METHODS = ('GET', 'POST', 'PUT', 'DELETE') class RestEndpoint: def __init__(self): self.methods = <> for method_name in DEFAULT_METHODS: method = getattr(self, method_name.lower(), None) if method: self.register_method(method_name, method) def register_method(self, method_name, method): self.methods[method_name.upper()] = method async def dispatch(self, request: Request): method = self.methods.get(request.method.upper()) if not method: raise HTTPMethodNotAllowed('', DEFAULT_METHODS) wanted_args = list(inspect.signature(method).parameters.keys()) available_args = request.match_info.copy() available_args.update('request': request>) unsatisfied_args = set(wanted_args) - set(available_args.keys()) if unsatisfied_args: # Expected match info that doesn't exist raise HttpBadRequest('') return await method(**arg_name: available_args[arg_name] for arg_name in wanted_args>) class CollectionEndpoint(RestEndpoint): def __init__(self, resource): super().__init__() self.resource = resource async def get(self) -> Response: data = [] notes = session.query(Note).all() for instance in self.resource.collection.values(): data.append(self.resource.render(instance)) data = self.resource.encode(data) return Response ( status=200, body=self.resource.encode( 'notes': [ 'id': note.id, 'title': note.title, 'description': note.description, 'created_at': note.created_at, 'created_by': note.created_by, 'priority': note.priority> for note in session.query(Note) ] >), content_type='application/json') async def post(self, request): data = await request.json() note=Note(title=data['title'], description=data['description'], created_at=data['created_at'], created_by=data['created_by'], priority=data['priority']) session.add(note) session.commit() return Response(status=201, body=self.resource.encode( 'notes': [ 'id': note.id, 'title': note.title, 'description': note.description, 'created_at': note.created_at, 'created_by': note.created_by, 'priority': note.priority> for note in session.query(Note) ] >), content_type='application/json') class InstanceEndpoint(RestEndpoint): def __init__(self, resource): super().__init__() self.resource = resource async def get(self, instance_id): instance = session.query(Note).filter(Note.id == instance_id).first() if not instance: return Response(status=404, body=json.dumps('not found': 404>), content_type='application/json') data = self.resource.render_and_encode(instance) return Response(status=200, body=data, content_type='application/json') async def put(self, request, instance_id): data = await request.json() note = session.query(Note).filter(Note.id == instance_id).first() note.title = data['title'] note.description = data['description'] note.created_at = data['created_at'] note.created_by = data['created_by'] note.priority = data['priority'] session.add(note) session.commit() return Response(status=201, body=self.resource.render_and_encode(note), content_type='application/json') async def delete(self, instance_id): note = session.query(Note).filter(Note.id == instance_id).first() if not note: abort(404, message="Note <> doesn't exist".format(id)) session.delete(note) session.commit() return Response(status=204) class RestResource: def __init__(self, notes, factory, collection, properties, id_field): self.notes = notes self.factory = factory self.collection = collection self.properties = properties self.id_field = id_field self.collection_endpoint = CollectionEndpoint(self) self.instance_endpoint = InstanceEndpoint(self) def register(self, router: UrlDispatcher): router.add_route('*', '/'.format(notes=self.notes), self.collection_endpoint.dispatch) router.add_route('*', '//>'.format(notes=self.notes), self.instance_endpoint.dispatch) def render(self, instance): return OrderedDict((notes, getattr(instance, notes)) for notes in self.properties) @staticmethod def encode(data): return json.dumps(data, indent=4).encode('utf-8') def render_and_encode(self, instance): return self.encode(self.render(instance)) 

By using async keyword with all methods (GET, POST, PUT and DELETE), we ensure that those operations are performed asynchronously and the response is returned from from both collection end point and instance end points. After setting up our endpoints, we declare resources in aio-app.py file.

from aiohttp.web import Application, run_app from aiohttp_rest import RestResource from models import Note from sqlalchemy import engine_from_config notes = <> app = Application() person_resource = RestResource('notes', Note, notes, ('title', 'description', 'created_at', 'created_by', 'priority'), 'title') person_resource.register(app.router) if __name__ == '__main__': run_app(app) 

Источник