aiohttp를 사용하여 REST API를 만드는 방법

aiohttp은 python 및 AsyncIO에 대한 HTTP 클라이언트/서버입니다. 서버 웹 소켓과 클라이언트 웹 소켓을 모두 지원합니다. 비동기식으로 작동하기 때문에 초당 수백 건의 요청을 처리할 수 있어 다른 프레임워크보다 더 나은 성능을 제공합니다.

AsyncIO은 쓰기를 위한 python 라이브러리입니다.

코루틴을 사용하는 단일 스레드 동시 코드.

소켓 및 기타 리소스를 통한 멀티플렉싱 I/O 액세스.

실행 중인 네트워크 클라이언트와 서버 및 기타 관련 프리미티브.

이는 특히 소켓 및 기타 리소스를 통한 I/O 바운드 작업에 대한 동시성을 제공합니다. 동시성은 사용자가 I/O 바운드 결과를 기다리지 않도록 합니다.



이 기사에서는 aiohttp를 사용하여 애플리케이션에 대한 나머지 API를 작성합니다. 메모 테이블이 있는 간단한 응용 프로그램입니다.

aiohttp 설정

Python 3에서 가상 환경을 활성화하고 aiohttp를 설치합니다.

pip install aiohttp

또는 github 저장소를 복제하고 요구 사항을 설치하십시오.

pip install -r requirements.txt

모델 만들기

models.py에서 sqlite를 데이터베이스로 사용하도록 애플리케이션을 구성합니다.

# 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()

그런 다음 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

자원

aiohttp_rest.py 파일에 API 끝점을 정의합니다.

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('*', '/{notes}'.format(notes=self.notes), self.collection_endpoint.dispatch)
        router.add_route('*', '/{notes}/{{instance_id}}'.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))

모든 메서드(GET, POST, PUT 및 DELETE)와 함께 async 키워드를 사용하여 이러한 작업이 비동기적으로 수행되고 응답이 수집 끝점과 인스턴스 끝점 모두에서 반환되도록 합니다. 엔드포인트를 설정한 후 aio-app.py 파일에서 리소스를 선언합니다.

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)

애플리케이션 실행

먼저 다음을 수행하여 데이터베이스를 생성합니다.

python models.py

터미널에서 다음을 실행하여 앱을 실행합니다.

python aio-app.py

파이썬 셸을 열고 일부 요청을 실행합니다.

requests.post('http://localhost:8080/notes',
                 data=json.dumps({ "title": "note two",
                 "created_at": "2017-08-23 00:00", "created_by": "apcelent", "description": "sample notes", "priority": 4
}))

requests.put('http://localhost:8080/notes/1',
                 data=json.dumps({ "title": "note edit",
                 "created_at": "2017-08-23 00:00", "created_by": "apcelent", "description": "sample notes edit", "priority": 4
}))


requests.delete('http://localhost:8080/notes/1')

이들은 aiohttp REST API를 사용하여 데이터베이스에 일부 메모를 생성합니다. 이 메모는 http://127.0.0.1:8080/notes에서 볼 수 있습니다.

소스 코드는 here에서 찾을 수 있습니다.

이 기사는 원래 Apcelent Tech Blog에 게재되었습니다.